package controllers import ( "apiserver/backend" "apiserver/common" "apiserver/models" "apiserver/services" "apiserver/store" "apiserver/utils" logger "apiserver/utils" "crypto/md5" "encoding/json" "errors" "fmt" "net/url" "os" "path/filepath" "regexp" "strconv" "strings" "time" "github.com/gin-gonic/gin" "github.com/google/uuid" "go.uber.org/zap" ) const ( //TODO: Need assign clusterID globally. clusterID = 1 apiserverID = 0 ) var ( SignalMap = map[string]string{ "kill": "KILL", "resume": "CONT", "stop": "STOP", } realCopyLocalFiles []string fakeCopyRemoteFile []string ) type JobController struct{} // GetJobsInfo godoc // @Summary Get jobs information // @Description Get jobs information // @Tags jobs // @Accept json // @Produce json // @Param filter query string "filter=[{user:eq:user1},{queue:eq:queue1}]" // @Param fields query string "fields=jobid,status,queue" // @Success 200 {array} models.JobInfo "Response for jobs info" // @Failure 400 {object} models.Error // @Router /api/v1/jobs [get] func (j JobController) GetJobsInfo(ctx *gin.Context) { logger.Debug("Enter GetJobsInfo() function from ", zap.String("client_ip", ctx.ClientIP())) var fields []string filters := make([]utils.FilterCondition, 0) sortOrders := make([]utils.SortOrder, 0) token := ctx.GetHeader("Authorization") submitUser, err := GetUsernameFromToken(token) if err != nil { ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } if ctx.IsAborted() { ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } // ---- check parameters ---- for param := range ctx.Request.URL.Query() { logger.Debug("GetJobsInfo param is ", zap.Any("param", param)) if !utils.ContainsField(utils.SupportedQueryParams, param) && !utils.ContainsField(utils.TypeQueryParams, param) { err := errors.New("invalid job 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=job_id:asc,submit_time:desc limitStr := ctx.Query("limit") // limit=100 pagetokenStr := ctx.Query("page_token") // page_token=Aafjakjf0023A typeStr := ctx.Query("type") // type=array_stats logger.Debug("Fields, filters values are:", zap.String("filterStr", filterStr), zap.String("fieldsStr", fieldsStr), zap.String("sortStr", sortStr), zap.String("limitStr", limitStr), zap.String("pagetokenStr", pagetokenStr), zap.String("typeStr", typeStr)) // ---- 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 type ---- typeValue, err := utils.ParseType(typeStr, models.JobAvailableTypes) 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 } } else { logger.Debug("query filter is not specified, set submit_user as default filter: filter=[{submit_user:eq:" + submitUser) filters = append(filters, utils.FilterCondition{ Field: "submit_user", Operator: "eq", Value: submitUser, }) } permission, err := GetRequireResourcePermission(submitUser, "queue:view") if err != nil { ReturnError(ctx, common.ERROR_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } var accessQueues []string if permission.DomainType == models.DomainTypeAll { accessQueues = store.GetAllQueueNames() } else if permission.DomainType == models.DomainTypeQueue { accessQueues = permission.DomainValues } logger.Debug("get job info accessQueues", zap.Any("accessQueues", accessQueues), zap.Any("permission", permission)) // check job:view permission permission, err = GetRequireResourcePermission(submitUser, "job:view") logger.Debug("get job info for job:view permission:", zap.Any("permission", permission)) if err != nil || len(accessQueues) == 0 { ReturnError(ctx, common.ERROR_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } // if submit_user is not specified, use query request submission user var specify_user bool = false var delindex int = -1 for i, filter := range filters { if filter.Field == "submit_user" { specify_user = true //handle submit_user = "all" if filter.Value == models.DomainTypeAll { delindex = i } if permission.DomainType != models.DomainTypeSelf { submitUser = filter.Value } break } } if delindex != -1 { filters = append(filters[:delindex], filters[delindex+1:]...) } if !specify_user || permission.DomainType == models.DomainTypeSelf { logger.Debug("submit_user is not specified in query filter, add submit_user into filter: filter=[{submit_user:eq:" + submitUser) filters = append(filters, utils.FilterCondition{ Field: "submit_user", Operator: "eq", Value: submitUser, }) } // job_id:eq:101, [array_idx:btw:(1,3), array_idx:btw:(6,8) ... ] compfilters := make(utils.CompFilterCondition) //filtercds := make([]utils.Condition, len(filters)) //subFilters is used to store internal filter extract from job_id:in:(101[1-2, 6-10], 102[1-2, 6-10]) //will be placed in a OR tree. for i, filter := range filters { // check the filter is supported. if typeValue == "array_stats" { if !utils.ContainsField(models.ARRAY_JOB_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 } } else { if !utils.ContainsField(models.JOB_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 } } // check whether exist queue_name for rbac if filter.Field == "queue_name" { if accessQueues != nil { found := false for _, q := range accessQueues { if filter.Value == q { found = true break } } if !found { logger.Debug("The queue name does not exist in accessible queues", zap.Any("filter.Value", filter.Value)) } } } // special handle job_name[index] cases, here index can be [1-3], [1,3,5],[1-3,5-7] situations if filters[i].Field == "job_name" { filtCdArrays := make([]utils.FilterConditionArray, 0) jobNames := extractJobNames(filters[i].Value) for i := range jobNames { jobNames[i] = strings.TrimSpace(jobNames[i]) if typeValue == "array_stats" { if strings.Contains(jobNames[i], "[") || strings.Contains(jobNames[i], "]") { err := errors.New("job_name with array index is not supported in statistics type") logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } } } logger.Debug("After extractJobNames jobNames is", zap.Any("jobNames", jobNames)) // only "in", "between", "notin" support multiple jobnames (include array job name) if !(filters[i].Operator == "in" || filters[i].Operator == "between" || filters[i].Operator == "notin") && len(jobNames) > 1 { err := fmt.Errorf("operator doesn't allow to use multiple values: %v", jobNames) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } if !(strings.Contains(filters[i].Value, "[") && strings.Contains(filters[i].Value, "]")) { continue } if strings.Contains(filters[i].Value, "[") || strings.Contains(filters[i].Value, "]") { if filters[i].Operator == "ne" { err := fmt.Errorf("operator doesn't support %v", jobNames) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } } for _, jobname := range jobNames { subORFilters := make([]utils.FilterCondition, 0) jobnameArry, err := parseArrayName(jobname) logger.Debug("After parseArrayName jobname is ", zap.Any("jobname", jobnameArry)) if err != nil { logger.Error("job name format is invalid", zap.Error(err)) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } if err = isValidArrayName(jobnameArry.KeyID); err != nil { err := errors.New("bad jobname: " + jobnameArry.KeyID) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } var operator string if filters[i].Operator == "in" { operator = "eq" } else { operator = filters[i].Operator } // treat every section as an OR condition mainFiltCond := utils.FilterCondition{ Field: filters[i].Field, Operator: operator, Value: jobnameArry.KeyID, } if jobnameArry.Indices != nil { // array jobid: job_id:in:(101[1-3, 5-6]) // handle every section in Indices list. for _, jidxsec := range jobnameArry.Indices { st, ed, err := parseArrayIndexSection(jidxsec.Index) if err != nil { err := errors.New("bad job array index: " + jidxsec.Index) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } if jidxsec.Step == 1 { subORFilters = append(subORFilters, utils.FilterCondition{ Field: "array_idx", Operator: "between", Value: "(" + strconv.Itoa(st) + "," + strconv.Itoa(ed) + ")", }) } else { // handle step > 1 idxList := make([]string, 0) for idx := st; idx <= ed; idx += jidxsec.Step { idxList = append(idxList, strconv.Itoa(idx)) } subORFilters = append(subORFilters, utils.FilterCondition{ Field: "array_idx", Operator: "in", Value: "(" + strings.Join(idxList, ",") + ")", }) } } } filtCdArrays = append(filtCdArrays, utils.FilterConditionArray{ MainFiltCond: mainFiltCond, SubFiltConds: subORFilters, }) } compfilters["job_name"] = filtCdArrays } if filters[i].Field == "job_id" { filtCdArrays := make([]utils.FilterConditionArray, 0) jobids := extractJobNames(filters[i].Value) for i := range jobids { jobids[i] = strings.TrimSpace(jobids[i]) if typeValue == "array_stats" { if strings.Contains(jobids[i], "[") || strings.Contains(jobids[i], "]") { err := errors.New("job_id with array index is not supported in statistics type") logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } } } logger.Debug("After extractJobNames jobids is", zap.Any("jobids", jobids)) if !(filters[i].Operator == "in" || filters[i].Operator == "between" || filters[i].Operator == "notin") && len(jobids) > 1 { err := fmt.Errorf("operator doesn't allow to use multiple values: %v", jobids) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } if !(strings.Contains(filters[i].Value, "[") && strings.Contains(filters[i].Value, "]")) { continue } if strings.Contains(filters[i].Value, "[") || strings.Contains(filters[i].Value, "]") { if filters[i].Operator == "ne" { err := fmt.Errorf("operator doesn't support %v", jobids) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } } for _, jobid := range jobids { subORFilters := make([]utils.FilterCondition, 0) jobidArry, err := parseArrayName(jobid) logger.Debug("After parseArrayName jobid is ", zap.Any("jobid", jobidArry)) if err != nil { logger.Error("job id format is invalid", zap.Error(err)) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } if !utils.IsNumeric(jobidArry.KeyID) { err := errors.New("bad jobid: " + jobidArry.KeyID) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } var operator string if filters[i].Operator == "in" { operator = "eq" } else { // for other operator, only one job is supported. // means job_id:eq:100[1-3,6-10]. // if job_id:eq:(100[1-3],101[1-2]) should be rejected // TODO: need check if already handle this case ??? operator = filters[i].Operator } // single jobid: job_id:in:(100, 101[1-3, 5-6]) // here, handle the job_id = 100 mainFiltCond := utils.FilterCondition{ Field: filters[i].Field, Operator: operator, Value: jobidArry.KeyID, } // treat every section as an OR condition if jobidArry.Indices != nil { // handle every section in Indices list. for _, jidxsec := range jobidArry.Indices { st, ed, err := parseArrayIndexSection(jidxsec.Index) if err != nil { err := errors.New("bad job array index: " + jidxsec.Index) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } if jidxsec.Step == 1 { subORFilters = append(subORFilters, utils.FilterCondition{ Field: "array_idx", Operator: "between", Value: "(" + strconv.Itoa(st) + "," + strconv.Itoa(ed) + ")", }) } else { // handle step > 1 idxList := make([]string, 0) for idx := st; idx <= ed; idx += jidxsec.Step { idxList = append(idxList, strconv.Itoa(idx)) } subORFilters = append(subORFilters, utils.FilterCondition{ Field: "array_idx", Operator: "in", Value: "(" + strings.Join(idxList, ",") + ")", }) } } } filtCdArrays = append(filtCdArrays, utils.FilterConditionArray{ MainFiltCond: mainFiltCond, SubFiltConds: subORFilters, }) } // end of for jobids compfilters["job_id"] = filtCdArrays } // filter with arrayjob[] // ==================== exec_hosts ================ // filter is exec_hosts need special handling // the exec_hosts query request is array, eg: "{exec_hosts:eq:h1,h2,hg1} if filters[i].Field == "exec_hosts" { if filters[i].Operator != "eq" { err := fmt.Errorf("filter exec_hosts doesn't support operator rather than eq") logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } hosts := strings.Split(strings.TrimSpace(filters[i].Value), ",") execHosts, err := GetHostList(hosts) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } //change the filter to get unique_job_id list from db. subFilters := make([]utils.FilterCondition, 0) subFilters = append(subFilters, utils.FilterCondition{ Field: "host_name", Operator: "in", Value: "(" + strings.Join(execHosts, ",") + ")", }) uniqueJobIds, err := FetchJobIdList(subFilters) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } if len(uniqueJobIds) == 0 { logger.Debug("no job found") ReturnSuccess(ctx, common.SUCCESS_CODE, common.SUCCESS_MSG, map[string]interface{}{ "count": 0, "rows": []interface{}{}, "page_token": "", }) return } //Replace filter condition from "exec_hosts" to "unique_job_id". uniqueJobIdStrs := make([]string, len(uniqueJobIds)) for i, uniqueJobId := range uniqueJobIds { uniqueJobIdStrs[i] = strconv.FormatInt(uniqueJobId, 10) } filters[i].Field = "unique_job_id" filters[i].Operator = "in" filters[i].Value = "(" + strings.Join(uniqueJobIdStrs, ",") } // end of "exec_hosts" filter } // end of for { filters } // remove filter with jobarray[] from filters, for filter in compfilters need special handling. if len(compfilters) > 0 { var filtersNew []utils.FilterCondition for _, filter := range filters { found := false for k, _ := range compfilters { if filter.Field == k { found = true break } } if found { continue } filtersNew = append(filtersNew, filter) } filters = filtersNew } var job interface{} if typeValue == "array_stats" { job = &models.ArrayJobInfo{} } else { job = &models.JobInfo{} } // Apply RBAC queue filter exactly once after the filter loop filters = applyQueueRBACFilter(filters, accessQueues) logger.Debug("before call ConvertToNewFilter for all jobs", zap.Any("filters", filters)) filtercds, err := utils.ConvertToNewFilter(filters, job) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } // convert filterConditionArray to compound conditions logger.Debug("before call ConvertToNewFilterArray for all array jobs", zap.Any("filtersarray", compfilters)) compfiltercd, err := utils.ConvertToNewFilterArray(compfilters, job) 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 typeValue == "array_stats" { fields = models.ARRAY_JOB_DEFAULT_COLUMN } else { if fieldsStr != "" { fields = utils.ParseFields(fieldsStr) found1 := false found2 := false found3 := false // verify the fields is in supported list supportedFields := utils.GetSupportedFields(models.JobInfo{}) utils.AddFieldsToSupportedList(&supportedFields, models.JobAdditionalFields{}) logger.Debug("job 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 job_id in fields, add job_id as default field. if field == "job_id" { found1 = true } if field == "unique_job_id" { found2 = true } if field == "job_name" { found3 = true } } if !found1 { logger.Debug("user doesn't specify job_id in query fields, add it.") fields = append(fields, "job_id") } if !found2 { fields = append(fields, "unique_job_id") } if found3 { fields = append(fields, "array_idx") } } else { fields = models.JOB_DEFAULT_COLUMN } } // Map fields to database fields for i, field := range fields { fields[i] = utils.GetMappedField(field) } // add some fields for rbac if typeValue == "array_stats" { fields = append(fields, "submit_user") } else { fields = append(fields, "submit_user", "queue_name") } // ---- check sort ---- // handle sort, when no sortStr specified, need add the primary key of job_info // table into sortStr as default value // parse sortStr var sorts []utils.SortOrder found := false if sortStr != "" { if typeValue == "array_stats" { sorts, err = utils.ParseSortParam(sortStr, models.ARRAY_JOB_SUPPORT_FILTERS) } else { sorts, err = utils.ParseSortParam(sortStr, models.JOB_SUPPORT_SORT) } if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } //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 } //the primary key of job_info table job_id should always be added into sortStr if user doesn't specify. if typeValue == "array_stats" { for _, sortent := range sorts { if sortent.Column == "job_id" { found = true break } } } else { for _, sortent := range sorts { if sortent.Column == "unique_job_id" { found = true break } } } } // user not specify the unique_job_id, just append "unique_job_id:asc" at the end. if !found || sortStr == "" { if typeValue == "array_stats" { logger.Debug("job_id is not specified, set sort=job_id:asc as default") sorts = append(sorts, utils.SortOrder{ Column: "job_id", Ascending: true, }) } else { logger.Debug("unique_job_id is not specified, set sort=unique_job_id:asc as default") sorts = append(sorts, utils.SortOrder{ Column: "unique_job_id", Ascending: true, }) } } sortOrders = sorts // ---- 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 } } queryOpts, err := utils.BuildQueryOptions(filtercds, compfiltercd, sortOrders, limit, pagetokenStr) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } logger.Debug("the job query request:", zap.Any("jquery request", filtercds), zap.Any("jquery request arrayjob:", compfiltercd), zap.Any("sortOrders", sortOrders), zap.Any("limit", limit), zap.String("pagetoken", pagetokenStr), zap.Any("fields", fields)) var jqueryReply []utils.OrderMap // Declare jqueryReply pagetoken := "" // Initialize pagetoken with an empty string or appropriate value if typeValue == "array_stats" { jqueryReply, pagetoken, err = services.QueryData(services.TableArrayJob, queryOpts, fields) } else { jqueryReply, pagetoken, err = services.QueryData(services.TableJobInfo, queryOpts, fields) } if err != nil { logger.Error("Failed to query job.", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } logger.Debug("the job info query reply:", zap.Any("jqueryReply", jqueryReply), zap.String("pagetoken", pagetoken)) //convert job_status to readable string. queryResult := processJobData(jqueryReply, pagetoken) ReturnSuccess(ctx, common.SUCCESS_CODE, common.SUCCESS_MSG, queryResult) } // GetJobByIDs godoc // @Summary Get multiple jobs by their IDs // @Description Retrieve information for multiple jobs in a single request with optional field selection // @Tags jobs // @Accept json // @Produce json // @Param jobids path string true "Comma-separated list of job IDs (e.g., 1001,1002,1003)" // @Param fields query string false "Comma-separated list of fields to return (e.g., jobid,status,queue)" // @Success 200 {object} JobsResponse "Successfully retrieved jobs" // @Failure 400 INVALID_PARAMS_CODE "invalid parameters" // @Failure 500 ERROR_MSG // @Example {curl} Example Usage: // @Example curl -X GET "http://localhost:8080/jobs/1001,1002,1003?fields=jobid,status,queue" // @Router /jobs/{jobids} [get] func (j JobController) GetJobByIDs(ctx *gin.Context) { logger.Debug("Enter GetJobByIDs() function from ", zap.String("client_ip", ctx.ClientIP())) token := ctx.GetHeader("Authorization") logonUser, err := GetUsernameFromToken(token) if err != nil { ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } if ctx.IsAborted() { ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } // ---- check parameters ---- for param := range ctx.Request.URL.Query() { logger.Debug("GetJobByIDs param is ", zap.Any("param", param)) if !utils.ContainsField(utils.SpecificSupportedQueryParams, param) && !utils.ContainsField(utils.TypeQueryParams, param) { err := errors.New("invalid specific job info query parameters of : " + param) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } } // Get job IDs from URL parameter jobIDsStr := ctx.Param("jobids") fieldsStr := ctx.Query("fields") typeStr := ctx.Query("type") // type=array_stats logger.Debug("GetJobByIDs origin query string jobid, Fields, filters values are:", zap.String("jobid", jobIDsStr), zap.String("fieldsStr", fieldsStr), zap.String("typeStr", typeStr)) // ---- check type ---- typeValue, err := utils.ParseType(typeStr, models.JobAvailableTypes) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, err.Error(), nil) return } // Extract job IDs using the new function jobids := extractJobNames(jobIDsStr) for i := range jobids { jobids[i] = strings.TrimSpace(jobids[i]) } logger.Debug("The input jobids", zap.Any("jobids", jobids)) mergeJqueryReply := make([]utils.OrderMap, 0) var page_token = "" for _, jobid := range jobids { compfilters := make(utils.CompFilterCondition) filtCdArrays := make([]utils.FilterConditionArray, 0) subORFilters := make([]utils.FilterCondition, 0) jobid = strings.TrimSpace(jobid) filters := make([]utils.FilterCondition, 0) sortOrders := make([]utils.SortOrder, 0) var err error // parse the jobid to support array list format jobid, err := parseArrayName(jobid) if err != nil { logger.Error("job id format is invalid", zap.Error(err)) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } logger.Debug("GetJobByIDs after parseArrayName", zap.Any("jobid", jobid)) if !utils.IsNumeric(jobid.KeyID) { err := errors.New("bad jobid: " + jobid.KeyID) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } filters = append(filters, utils.FilterCondition{ Field: "job_id", Operator: "eq", Value: jobid.KeyID, }) for _, filter := range filters { // check the filter is supported. if typeValue == "array_stats" { if !utils.ContainsField(models.ARRAY_JOB_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 } } else { if !utils.ContainsField(models.JOB_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 } } } mainFiltCond := utils.FilterCondition{ Field: filters[0].Field, Operator: filters[0].Operator, Value: jobid.KeyID, } // treat every section as an OR condition if jobid.Indices != nil && typeValue != "array_stats" { // handle every section in Indices list. for _, jidxsec := range jobid.Indices { st, ed, err := parseArrayIndexSection(jidxsec.Index) if err != nil { err := errors.New("bad job array index: " + jidxsec.Index) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } if jidxsec.Step == 1 { subORFilters = append(subORFilters, utils.FilterCondition{ Field: "array_idx", Operator: "between", Value: "(" + strconv.Itoa(st) + "," + strconv.Itoa(ed) + ")", }) } else { // handle step > 1 idxList := make([]string, 0) for idx := st; idx <= ed; idx += jidxsec.Step { idxList = append(idxList, strconv.Itoa(idx)) } subORFilters = append(subORFilters, utils.FilterCondition{ Field: "array_idx", Operator: "in", Value: "(" + strings.Join(idxList, ",") + ")", }) } } } filtCdArrays = append(filtCdArrays, utils.FilterConditionArray{ MainFiltCond: mainFiltCond, SubFiltConds: subORFilters, }) compfilters["job_id"] = filtCdArrays // convert filterConditionArray to compound conditions logger.Debug("before call ConvertToNewFilterArray for all jobs", zap.Any("compfilters", compfilters)) var job interface{} if typeValue == "array_stats" { job = &models.ArrayJobInfo{} } else { job = &models.JobInfo{} } compfiltercd, err := utils.ConvertToNewFilterArray(compfilters, job) 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 typeValue == "array_stats" { fields = models.ARRAY_JOB_DEFAULT_COLUMN } else { if fieldsStr != "" { fields = utils.ParseFields(fieldsStr) found1 := false found2 := false // verify the fields is in supported list supportedFields := utils.GetSupportedFields(models.JobInfo{}) utils.AddFieldsToSupportedList(&supportedFields, models.JobAdditionalFields{}) logger.Debug("job 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 job_id in fields, add job_id as default field. if field == "job_id" { found1 = true } if field == "unique_job_id" { found2 = true } } if !found1 { logger.Debug("user doesn't specify job_id in query fields, add it.") fields = append(fields, "job_id") } if !found2 { fields = append(fields, "unique_job_id") } } else { // default display columns fields = models.JOB_DEFAULT_COLUMN } } // Map fields to database fields for i, field := range fields { fields[i] = utils.GetMappedField(field) } // add some fields for rbac if typeValue == "array_stats" { fields = append(fields, "submit_user") } else { fields = append(fields, "submit_user", "queue_name") } // add a default order: sort=job_id:asc if typeValue == "array_stats" { sortOrders = append(sortOrders, utils.SortOrder{ Column: "job_id", // use unique_job_id as primary key Ascending: true, }) } else { sortOrders = append(sortOrders, utils.SortOrder{ Column: "unique_job_id", // use unique_job_id as primary key Ascending: true, }) } var bookmark string = "" var limit int = utils.PER_PAGE_LIMIT_MAX for { queryOpts, err := utils.BuildQueryOptions(nil, compfiltercd, sortOrders, limit, bookmark) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } logger.Debug("the job query request:", zap.Any("jquery request arrayjob:", compfiltercd), zap.Any("sortOrders", sortOrders), zap.Any("limit", limit), zap.String("pagetoken", bookmark), zap.Any("fields", fields)) var jqueryReply []utils.OrderMap // Declare jqueryReply if typeValue == "array_stats" { jqueryReply, page_token, err = services.QueryData(services.TableArrayJob, queryOpts, fields) } else { jqueryReply, page_token, err = services.QueryData(services.TableJobInfo, queryOpts, fields) } if err != nil { logger.Error("Failed to query job info.", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } // Here, page_token is escaped and will not be recognized by dbapi, // for internal use, need add Unescape here. bookmark, _ = url.QueryUnescape(page_token) mergeJqueryReply = append(mergeJqueryReply, jqueryReply...) if bookmark == "" { break } } // end for } // end for jobid logger.Debug("the specific jobid query reply after merge:", zap.Any("jqueryReply", mergeJqueryReply)) // check rbac removeFlag := false for i, row := range mergeJqueryReply { jobPermission, err := GetRequireResourcePermission(logonUser, "job:view") if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } queuePermission, err := GetRequireResourcePermission(logonUser, "queue:view") if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } if job_owner, exist := row.Data["submit_user"]; exist { if jobPermission.DomainType == models.DomainTypeSelf { if logonUser != job_owner.(string) { removeFlag = true mergeJqueryReply = append(mergeJqueryReply[:i], mergeJqueryReply[i+1:]...) } } } if typeValue != "array_stats" { if job_queue, exist := row.Data["queue_name"]; exist { queueName, _ := job_queue.(string) if queuePermission.DomainType == models.DomainTypeQueue { permissionQueueValues := queuePermission.DomainValues if !utils.ContainsField(permissionQueueValues, queueName) { removeFlag = true mergeJqueryReply = append(mergeJqueryReply[:i], mergeJqueryReply[i+1:]...) } } } } } if removeFlag && len(mergeJqueryReply) == 0 { ReturnError(ctx, common.ERROR_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } //convert job_status to readable string. queryResult := processJobData(mergeJqueryReply, page_token) ReturnSuccess(ctx, common.SUCCESS_CODE, common.SUCCESS_MSG, queryResult) } func extractJobNames(jobNamesStr string) []string { // Trim surrounding parentheses if present if len(jobNamesStr) > 1 && jobNamesStr[0] == '(' && jobNamesStr[len(jobNamesStr)-1] == ')' { jobNamesStr = jobNamesStr[1 : len(jobNamesStr)-1] } var jobNames []string currentJobName := "" bracketCount := 0 for _, char := range jobNamesStr { if char == '[' { bracketCount++ } else if char == ']' { bracketCount-- } else if char == ',' && bracketCount == 0 { // If we encounter a comma and we're not inside brackets, we split here jobNames = append(jobNames, currentJobName) currentJobName = "" continue } currentJobName += string(char) } // Add the last job name if there's any remaining if currentJobName != "" { jobNames = append(jobNames, currentJobName) } return jobNames } func (j JobController) RestartJobByID(ctx *gin.Context) { logger.Debug("Enter RestartJobByID() function from ", zap.String("client_ip", ctx.ClientIP())) // get submit user from token token := ctx.GetHeader("Authorization") restarter, err := GetUsernameFromToken(token) if err != nil { logger.Error("Get username from token error", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } // check jobid value jobIDStr := ctx.Param("jobids") logger.Debug("Receive job restart request:", zap.String("jobId", jobIDStr), zap.String("restarter", restarter)) if err = isValidJobId(jobIDStr); err != nil { logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } // check post body whether correct var req models.JobRestartRequest 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 } err = preCheckJobRestartRequest(&req) if err != nil { logger.Error("Failed to restart job", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } compfilters := make(utils.CompFilterCondition) filtCdArrays := make([]utils.FilterConditionArray, 0) subORFilters := make([]utils.FilterCondition, 0) jobid := strings.TrimSpace(jobIDStr) filters := make([]utils.FilterCondition, 0) sortOrders := make([]utils.SortOrder, 0) filters = append(filters, utils.FilterCondition{ Field: "job_id", Operator: "eq", Value: jobid, }) mainFiltCond := utils.FilterCondition{ Field: filters[0].Field, Operator: filters[0].Operator, Value: jobid, } fields := utils.GetSupportedFields(models.JobInfo{}) filtCdArrays = append(filtCdArrays, utils.FilterConditionArray{ MainFiltCond: mainFiltCond, SubFiltConds: subORFilters, }) compfilters["job_id"] = filtCdArrays var job interface{} job = &models.JobInfo{} compfiltercd, err := utils.ConvertToNewFilterArray(compfilters, job) var limit int = utils.PER_PAGE_LIMIT_MAX var bookmark string = "" queryOpts, err := utils.BuildQueryOptions(nil, compfiltercd, sortOrders, limit, bookmark) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } logger.Debug("the job query request:", zap.Any("jquery request arrayjob:", compfiltercd), zap.Any("sortOrders", sortOrders), zap.Any("limit", limit), zap.String("pagetoken", bookmark), zap.Any("fields", fields)) var jqueryReply []utils.OrderMap jqueryReply, page_token, err := services.QueryData(services.TableJobInfo, queryOpts, fields) if err != nil { logger.Error("Failed to query job info.", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } // Here, page_token is escaped and will not be recognized by dbapi, bookmark, _ = url.QueryUnescape(page_token) if len(jqueryReply) == 0 { logger.Error("No job found for the given job ID", zap.String("jobID", jobIDStr)) ReturnError(ctx, common.ERROR_NOT_FOUND_CODE, common.ERROR_JOB_NOT_FOUND_MSG, nil) return } oldJob := jqueryReply[0].Data // Get the first job info logger.Debug("Old job info:", zap.Any("oldJob", oldJob)) if v, ok := oldJob["submit_user"].(string); ok { if v != restarter { err := fmt.Errorf("job <%s> is not submitted by <%s>, cannot be restarted by <%s>", jobIDStr, v, restarter) logger.Error(err.Error()) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } } jobID, err := backend.GenerateJobID(clusterID, apiserverID, 1) if err != nil { logger.Error("Failed to generate job ID for restart", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, "failed to generate job ID: "+err.Error(), nil) return } // Use a concrete type for job newJob := models.JobInfo{ Unique_JobID: int64(jobID), JobID: int64(backend.ExtractArrayBaseID(jobID)), ArrayIdx: backend.ExtractArrayIdx(jobID), SubmitUser: restarter, JobStatus: models.JOBSTAT["PEND"], } err = processJobRestartRequest(&newJob, &req) if err != nil { logger.Error("Failed to restart job", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } if v, ok := oldJob["job_name"].(string); ok { newJob.JobName = v } if chkpntStr, ok := oldJob["chkpnt_info"].(string); ok && chkpntStr != "" { var chkpntInfo map[string]interface{} if err := json.Unmarshal([]byte(chkpntStr), &chkpntInfo); err == nil { if restartCmd, ok := chkpntInfo["restart_command"].(string); ok { logger.Debug("Restart job with checkpoint command", zap.String("restart_command", restartCmd)) newJob.Command = restartCmd } if ckpt_pid, ok := chkpntInfo["ckpt_pid"].(float64); ok { extendData := map[string]interface{}{} if v, ok := oldJob["extend_data"].(string); ok && v != "" { _ = json.Unmarshal([]byte(v), &extendData) } if v, ok := extendData["sub_env_vars"]; ok && v != nil { if sub, ok := v.(map[string]interface{}); ok { sub["XCE_JOB_CHECKPOINT_PID"] = strconv.FormatFloat(ckpt_pid, 'f', -1, 64) extendData["sub_env_vars"] = sub } } else { extendData["sub_env_vars"] = map[string]interface{}{ "XCE_JOB_CHECKPOINT_PID": strconv.FormatFloat(ckpt_pid, 'f', -1, 64), } } extendDataBytes, _ := json.Marshal(extendData) newJob.ExtendData = models.Jsonb{String: string(extendDataBytes)} } else { if v, ok := oldJob["extend_data"].(string); ok && v != "" { newJob.ExtendData = models.Jsonb{String: v} } } } } else { if v, ok := oldJob["command"].(string); ok { newJob.Command = v } } newJob.ChkpntInfo = models.Jsonb{} //Reset checkpoint_info field if v, ok := oldJob["queue_name"].(string); ok { newJob.QueueName = v } if v, ok := oldJob["project_name"].(string); ok { newJob.ProjectName = v } if v, ok := oldJob["user_group"].(string); ok { newJob.UserGroup = v } if v, ok := oldJob["job_group"].(string); ok { newJob.JobGroup = v } if v, ok := oldJob["license_project"].(string); ok { newJob.LicenseProject = v } if v, ok := oldJob["job_description"].(string); ok { newJob.JobDescription = v } if v, ok := oldJob["res_req"].(string); ok { newJob.ResReq = v } if v, ok := oldJob["rsv_id"].(string); ok { newJob.RsvID = v } if v, ok := oldJob["sla"].(string); ok { newJob.SLA = v } if v, ok := oldJob["maxnum_processors"].(int); ok { newJob.MaxNumProcessors = v } if v, ok := oldJob["num_processors"].(int); ok { newJob.NumProcessors = v } if newJob.AskedHosts == "" { if v, ok := oldJob["asked_hosts"].(string); ok { newJob.AskedHosts = v } } if v, ok := oldJob["begin_time"].(time.Time); ok { newJob.BeginTime = models.CustomTime{Time: v} } if v, ok := oldJob["term_time"].(time.Time); ok { newJob.TermTime = models.CustomTime{Time: v} } if v, ok := oldJob["job_priority"].(int); ok { newJob.JobPriority = v } if v, ok := oldJob["user_priority"].(int); ok { newJob.UserPriority = v } if v, ok := oldJob["depend_cond"].(string); ok { newJob.DependCond = v } if v, ok := oldJob["cwd"].(string); ok { newJob.CWD = v } if v, ok := oldJob["from_host"].(string); ok { newJob.FromHost = v } if v, ok := oldJob["pre_exec_cmd"].(string); ok { newJob.PreExecCmd = v } if v, ok := oldJob["post_exec_cmd"].(string); ok { newJob.PostExecCmd = v } if v, ok := oldJob["requeue_error_values"].(string); ok { newJob.RequeueErrorValues = v } if v, ok := oldJob["run_time_estimation"].(int); ok { newJob.RunTimeEstimation = v } if v, ok := oldJob["mig_threshold"].(int); ok { newJob.MigThreshold = v } if v, ok := oldJob["chkpnt_period"].(int); ok { newJob.ChkpntPeriod = v } if v, ok := oldJob["chkpnt_dir"].(string); ok { newJob.ChkpntDir = v } if v, ok := oldJob["out_dir"].(string); ok { newJob.OutDir = v } if v, ok := oldJob["in_file"].(string); ok { newJob.InFile = v } if v, ok := oldJob["err_file"].(string); ok { newJob.ErrFile = v } if v, ok := oldJob["out_file"].(string); ok { newJob.OutFile = v } if v, ok := oldJob["login_shell"].(string); ok { newJob.LoginShell = v } logger.Info("Generated job ID for restart", zap.Int64("jobID", newJob.JobID)) err = services.SubmitJob(&newJob, 1) if err != nil { logger.Error("Failed to submit job for restart process.", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil) return } response := models.JobSubmitResponse{ Message: fmt.Sprintf("Job <%s> will be restarted with new job <%d> to default queue <%s>.", jobid, newJob.JobID, newJob.QueueName), JobID: newJob.JobID, QueueName: newJob.QueueName, } ReturnSuccess(ctx, common.CREATED, common.SUCCESS_MSG, response) return } // CreateJob godoc // @Summary Create a new job // @Description Submit a new job with specified requirements and parameters // @Tags jobs // @Accept json // @Produce json // @Param job body JobSubmitRequest true "Job submission parameters" // @Success 201 {object} JobSubmitResponse "Job successfully created" // @Failure 400 {object} ErrorResponse "Invalid request body or parameters" // @Failure 401 {object} ErrorResponse "Unauthorized - Invalid or missing credentials" // @Failure 403 {object} ErrorResponse "Forbidden - User lacks permission" // @Failure 500 {object} ErrorResponse "Internal server error" // @Router /jobs [post] func (j JobController) CreateJob(ctx *gin.Context) { logger.Debug("Enter CreateJob() function from ", zap.String("client_ip", ctx.ClientIP())) token := ctx.GetHeader("Authorization") submitUser, err := GetUsernameFromToken(token) if err != nil { ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } var req models.JobSubmitRequest 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 } // precheck the request parameter before generating the job ID parsedIndices, err := preCheckJobRequest(&req) if err != nil { logger.Error("Failed to submit job request", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } // check rbac queuePermission, err := GetRequireResourcePermission(submitUser, "queue:submit") if err != nil { logger.Error("Permission check failed for queue", zap.String("queue", req.QueueName), zap.Error(err)) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } if queuePermission.DomainType == models.DomainTypeQueue { reqQueue := req.QueueName queuePermValues := queuePermission.DomainValues if !utils.ContainsField(queuePermValues, reqQueue) { logger.Error("Permission check failed for queue", zap.String("queue", req.QueueName), zap.Error(err)) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } } // Determine array size: 1 for normal jobs, >1 for array jobs arraySize := 1 if len(parsedIndices) > 1 { arraySize = len(parsedIndices) } // Generate job ID via C idgen (single call for both normal and array jobs) jobID, err := backend.GenerateJobID(clusterID, apiserverID, arraySize) if err != nil { logger.Error("Failed to generate job ID", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, "failed to generate job ID: "+err.Error(), nil) return } primaryJobID := int64(jobID) logger.Info("Generated job ID", zap.Int64("jobID", primaryJobID), zap.Int("arraySize", arraySize)) defaultValues := map[string]string{ "job_id": backend.FormatJobID(primaryJobID), "user_name": submitUser, "job_group": req.JobGroup, "project_name": req.ProjectName, } logger.Debug("realCopyLocalFiles=", zap.Any("realCopyLocalFiles", realCopyLocalFiles)) logger.Debug("fakeCopyRemoteFile=", zap.Any("fakeCopyRemoteFile", fakeCopyRemoteFile)) err = processJobRequest(ctx, &req, defaultValues) if err != nil { logger.Error("Failed to submit job request", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } // Build the single JobInfo to submit job := models.JobInfo{ Unique_JobID: primaryJobID, JobID: int64(backend.ExtractArrayBaseID(uint64(primaryJobID))), ArrayIdx: backend.ExtractArrayIdx(uint64(primaryJobID)), //SubmitTime: time.Now().Unix(), // backend will fill the submit time; SubmitUser: submitUser, JobStatus: models.JOBSTAT["PEND"], // new job is pend. JobName: req.JobName, Command: req.Command, QueueName: req.QueueName, //TODO: need set to DEFAULT_QUEUE ProjectName: req.ProjectName, //TODO: need set "default" if not specify? UserGroup: req.UserGroup, LicenseProject: req.LicenseProject, JobDescription: req.JobDescription, ResReq: req.Resreq, RsvID: req.RsvID, SLA: req.SLA, MaxNumProcessors: req.MaxNumProcessors, NumProcessors: req.NumProcessors, AskedHosts: req.AskedHosts, LaunchType: req.LaunchType, //BeginTime: req.BeginTime, //TermTime: req.TermTime, BeginTime: models.CustomTime{Time: req.BeginTime.Time}, TermTime: models.CustomTime{Time: req.TermTime.Time}, JobPriority: req.JobPriority, UserPriority: req.JobPriority, DependCond: req.DependCond, CWD: req.CWD, FromHost: req.FromHost, // bug-1676: client-asserted hostname, passed through unchanged PreExecCmd: req.PreExecCmd, PostExecCmd: req.PostExecCmd, RequeueErrorValues: req.RequeueErrorValues, RunTimeEstimation: req.RunTimeEstimation, MigThreshold: req.MigThreshold, ChkpntPeriod: req.ChkpntPeriod, ChkpntDir: req.ChkpntDir, OutDir: req.OutDir, InFile: req.InFile, // transfer file into json with absolute path ErrFile: req.ErrFile, OutFile: req.OutFile, LoginShell: req.LoginShell, ExtendData: models.Jsonb{String: req.ExtendData}, } err = postJobRequest(&job) if err != nil { logger.Error("Failed to handle postJobRequest ", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } logger.Debug("jobinfo before submit.", zap.Any("job", job)) // Submit job to backend (single call handles both normal and array) err = services.SubmitJob(&job, arraySize) if err != nil { logger.Error("Failed to submit job.", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil) return } logger.Info("Job submission successfully from ", zap.String("client_ip", ctx.ClientIP()), zap.Int64("jobID", primaryJobID)) response := models.JobSubmitResponse{ Message: fmt.Sprintf("Job <%d> is submitted to default queue <%s>.", primaryJobID, req.QueueName), JobID: primaryJobID, QueueName: req.QueueName, } ReturnSuccess(ctx, common.CREATED, common.SUCCESS_MSG, response) } func preCheckJobRestartRequest(req *models.JobRestartRequest) error { logger.Debug("Enter preCheckJobRestartRequest") if err := validateAskedHosts(req.AskedHosts); err != nil { return err } return nil } func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) { logger.Debug("Enter preCheckJobRequest") if req.QueueName == "" { req.QueueName = "normal" } else { // Reject submissions to the reserved internal recovery queue. if req.QueueName == "lost_and_found" { logger.Error("Job submission to reserved queue rejected", zap.String("queue_name", req.QueueName)) return nil, fmt.Errorf( "cannot submit jobs to 'lost_and_found' — this is an internal recovery queue") } //validate queue_name whether exists queueInfo, ok := store.Queues.Get(req.QueueName) if !ok { logger.Error("Invalid queue name specified", zap.String("queue_name", req.QueueName)) return nil, fmt.Errorf("invalid queue name: %s", req.QueueName) } queue_statusStr := formatQueueStatus(int64(queueInfo.QueueStatus)) if strings.HasPrefix(strings.ToLower(queue_statusStr), "closed") { logger.Error("Queue is closed", zap.String("queue_name", req.QueueName)) return nil, fmt.Errorf("cannot submit job to queue %s: queue is closed", req.QueueName) } } if req.JobName != "" { arrayName, idx, slotLimit, err := createArrayName(req.JobName) if err != nil { logger.Error("job name format is invalid", zap.Error(err)) return nil, errors.New("job array list format is invalid: " + err.Error()) } if arrayName != "" { if err = isValidArrayName(arrayName); err != nil { logger.Error(err.Error()) return nil, errors.New("job array list format is invalid: " + err.Error()) } } req.JobName = arrayName logger.Debug("preCheckJobRequest parsed job name is: ", zap.String("jobName", req.JobName), zap.Any("indices", idx), zap.Any("slotLimit", slotLimit)) return idx, nil } else { if req.Command != "" { if len(req.Command) > 4094 { req.JobName = utils.TruncateString(req.Command, 4094) } else { req.JobName = req.Command } } } if req.JobDescription != "" { if len(req.JobDescription) > 4094 { return nil, errors.New("job description length exceeds the limit of 4094 characters") } } // Validate AskedHosts using the new function if err := validateAskedHosts(req.AskedHosts); err != nil { return nil, err } if req.ProjectName != "" { if len(req.ProjectName) > 511 { return nil, errors.New("project name length exceeds the limit of 511 characters") } } if req.Resreq != "" { if len(req.Resreq) > 4094 { return nil, errors.New("resreq length exceeds the limit of 4094 characters") } } if req.ShellLimit.CPULimit != 0 { req.CPULimit.Seconds = (req.ShellLimit.CPULimit) / 1000 // convert ms to seconds } if req.ShellLimit.FileLimit.MB != 0 { req.FileLimit = req.ShellLimit.FileLimit } if req.ShellLimit.DataLimit.MB != 0 { req.DataLimit = req.ShellLimit.DataLimit } if req.ShellLimit.StackLimit.MB != 0 { req.StackLimit = req.ShellLimit.DataLimit } if req.ShellLimit.StackLimit.MB != 0 { req.StackLimit = req.ShellLimit.StackLimit } if req.ShellLimit.CoreLimit.MB != 0 { req.CoreLimit = req.ShellLimit.CoreLimit } if req.ShellLimit.ProcessLimit.MB != 0 { req.ProcessLimit = req.ShellLimit.ProcessLimit } if req.ShellLimit.ThreadLimit.MB != 0 { req.ThreadLimit = req.ShellLimit.ThreadLimit } if req.ShellLimit.MemLimit.MB != 0 { req.MemLimit = req.ShellLimit.MemLimit } if req.ShellLimit.SwapLimit.MB != 0 { req.SwapLimit = req.ShellLimit.SwapLimit } if req.ShellLimit.NumFiles != 0 { req.NumFiles = req.ShellLimit.NumFiles } // handle environment variables if req.Env != nil { if len(req.Env) > 4094 { return nil, errors.New("environment variable length exceeds the limit of 4094 characters") } for _, value := range req.Env { parseEnvs := strings.SplitN(value, "=", 2) if len(parseEnvs) != 2 { return nil, errors.New("environment variable should use key=value format") } key := parseEnvs[0] if strings.TrimSpace(key) == "" { return nil, errors.New("environment variable key cannot be empty") } } } if req.TransferFiles != nil { // handle transfer file and update InFile realCopyLocalFiles = make([]string, 0) fakeCopyRemoteFile = make([]string, 0) var extendData []map[string]string for _, transferFileStr := range req.TransferFiles { localFile, operator, remoteFile, err := parseTransferFiles(transferFileStr) if err != nil { return nil, err } /* based on different operator, check the local file and remote file >: copy local to remote <> and ><: copy local to remote <: copy remote to local <<: append remote to local */ if operator == ">" || operator == "<>" || operator == "><" { if localFile == "" { return nil, errors.New("local file is required for operator " + operator) } // Append to realCopyLocalFiles for later processing realCopyLocalFiles = append(realCopyLocalFiles, localFile) } if operator == "<" || operator == "<<" { if remoteFile == "" { return nil, errors.New("remote file is required for operator " + operator) } fakeCopyRemoteFile = append(fakeCopyRemoteFile, remoteFile) } // Add to inFileData extendData = append(extendData, map[string]string{ "local_file": localFile, "op": operator, "remote_file": remoteFile, }) } // Convert inFileData to JSON and assign to req.InFile extendDataJSON, err := json.Marshal(extendData) if err != nil { return nil, fmt.Errorf("failed to marshal InFile JSON: %v", err) } req.ExtendData = string(extendDataJSON) } if req.NumProcessors != 0 && req.MaxNumProcessors != 0 { minTasks := req.NumProcessors maxTasks := req.MaxNumProcessors minTasks, maxTasks, err := checkNumProcessors(minTasks, maxTasks) if err != nil { logger.Error("Failed to parse NumProcessors", zap.Any("numProcessors", req.NumProcessors), zap.Error(err)) return nil, fmt.Errorf("invalid NumProcessors specification: %w", err) } req.NumProcessors = minTasks req.MaxNumProcessors = maxTasks } else if req.NumProcessors != 0 { req.MaxNumProcessors = req.NumProcessors } if req.MigThreshold < 0 || req.MigThreshold > 2147483647 { return nil, errors.New("Incorrect migration threshold value. Specify a number greater than or equal to zero.") } if !utils.ContainsField(models.ValidInteractiveValues, strconv.Itoa(req.Interactive)) { return nil, errors.New("Interactive field must be 0, 1, or 2") } if req.LaunchType != "" { launchType := strings.ToLower(req.LaunchType) if !utils.ContainsField(models.ValidLaunchTypes, launchType) { return nil, errors.New("Launch_type must be openmpi or mpich") } req.LaunchType = launchType } return nil, nil } // It returns minTasks, maxTasks, and an error if parsing fails or constraints are violated. func checkNumProcessors(minTasks, maxTasks int) (minValue, maxValue int, err error) { logger.Debug("Enter checkNumProcessors", zap.Any("minTasks", minTasks), zap.Any("maxTasks", maxTasks)) if minTasks > maxTasks { err = fmt.Errorf("minTasks (%d) cannot be greater than maxTasks (%d)", minTasks, maxTasks) logger.Error("checkNumProcessors validation failed", zap.Error(err)) return 0, 0, err } return minTasks, maxTasks, nil } func parseArrayName(jobName string) (models.JobArrayInfo, error) { logger.Debug("Enter parseArrayName", zap.String("jobName", jobName)) var jobArrayInfo models.JobArrayInfo // Check if the jobName contains brackets if !strings.Contains(jobName, "[") { // If no brackets, return the job name as is jobArrayInfo.KeyID = jobName return jobArrayInfo, nil } openBracketIndex := strings.Index(jobName, "[") closeBracketIndex := strings.Index(jobName, "]") jobArrayInfo.KeyID = jobName[:openBracketIndex] if openBracketIndex != -1 && closeBracketIndex != -1 { if closeBracketIndex <= openBracketIndex { return models.JobArrayInfo{}, fmt.Errorf("invalid [] format") } indicesPart := jobName[openBracketIndex+1 : len(jobName)-1] // Remove the closing bracket // Parse the indices part rangeParts := strings.Split(indicesPart, ",") for _, part := range rangeParts { part = strings.TrimSpace(part) if strings.Contains(part, "-") { // Handle ranges like 1-3 bounds := strings.Split(part, "-") if len(bounds) != 2 { return models.JobArrayInfo{}, fmt.Errorf("invalid range format: %s", part) } step := 1 // Default step for ranges parts := strings.Split(part, ":") if len(parts) == 2 { var err error step, err = strconv.Atoi(parts[1]) if err != nil { return models.JobArrayInfo{}, fmt.Errorf("invalid step value: %v", err) } if step != 1 { return models.JobArrayInfo{}, fmt.Errorf("invalid step value and only support 1 now") } } part = parts[0] jobArrayInfo.Indices = append(jobArrayInfo.Indices, models.JobIndex{Index: part, Step: step}) } else { // Handle single indices step := 1 // Default step for single indices jobArrayInfo.Indices = append(jobArrayInfo.Indices, models.JobIndex{Index: part, Step: step}) } } } else { return models.JobArrayInfo{}, fmt.Errorf("no valid indices found in array job") } return jobArrayInfo, nil } // based on jobName to generate arrayName and individual indices func createArrayName(jobName string) (arrayName string, indices []int, slimit int, err error) { logger.Debug("Enter createArrayName", zap.Any("jobName=", jobName)) // find the index of the first '[' slotLimit := "" slotLimitValue := -1 openBracketIndex := strings.Index(jobName, "[") closeBracketIndex := strings.Index(jobName, "]") stepIndex := strings.Index(jobName, ":") slotLimitIndex := strings.Index(jobName, "%") logger.Debug("Debugging index values", zap.Any("openBracketIndex", openBracketIndex), zap.Any("closeBracketIndex", closeBracketIndex), zap.Any("stepIndex", stepIndex), zap.Any("slotLimitIndex", slotLimitIndex)) if slotLimitIndex != -1 { if slotLimitIndex+1 != len(jobName) { slotLimit = jobName[slotLimitIndex+1 : len(jobName)] } else { return "", nil, -1, fmt.Errorf("invalid slot limit value '%s': must be a positive integer", slotLimit) } logger.Debug("slotLimit=", zap.Any("slotLimit", slotLimit)) slotLimitValue, err = strconv.Atoi(strings.TrimSpace(slotLimit)) if err != nil || slotLimitValue < 1 { return "", nil, -1, fmt.Errorf("invalid slot limit value '%s': must be a positive integer", slotLimit) } } if openBracketIndex != -1 && closeBracketIndex != -1 { if closeBracketIndex <= openBracketIndex { return "", nil, -1, fmt.Errorf("invalidate job_name format") } if stepIndex != -1 && (stepIndex == closeBracketIndex-1 || stepIndex > closeBracketIndex) { return "", nil, -1, fmt.Errorf("invalidate step value") } // extract array_name arrayName = jobName[:openBracketIndex] // extract the part inside the brackets rangePart := jobName[openBracketIndex+1 : closeBracketIndex] // split the range parts by comma rangeParts := strings.Split(rangePart, ",") seenIndices := make(map[int]struct{}) // detect the duplicate element for _, part := range rangeParts { if part == "" { return "", nil, -1, fmt.Errorf("do not use empty range part: %s", part) } part = strings.TrimSpace(part) // remove spaces // Remove %slotLimitValue suffix if present if slotLimitValue != -1 { suffixToRemove := fmt.Sprintf("%%%d", slotLimitValue) part = strings.TrimSuffix(part, suffixToRemove) } if strings.Contains(part, "-") { // split by ':' to get the range and optional step parts := strings.Split(part, ":") if len(parts) < 1 || len(parts) > 2 { return "", nil, -1, fmt.Errorf("invalid range format: %s", part) } // parse the range bounds rangeBounds := strings.Split(parts[0], "-") if len(rangeBounds) != 2 { return "", nil, -1, fmt.Errorf("invalid range bounds: %s", part) } start, err := strconv.Atoi(strings.TrimSpace(rangeBounds[0])) if err != nil || start < 1 { return "", nil, -1, fmt.Errorf("invalid start value: %v, must be >= 1", err) } logger.Debug("start is:", zap.Any("start", start)) end, err := strconv.Atoi(strings.TrimSpace(rangeBounds[1])) if err != nil || end > utils.JobYamlConfig.MaxArraySize { return "", nil, -1, fmt.Errorf("invalid end value: %v, must be <= %d", err, utils.JobYamlConfig.MaxArraySize) } logger.Debug("end is:", zap.Any("end", end)) // set default step step := 1 if len(parts) == 2 { if stepIndex != -1 { step, err = strconv.Atoi(jobName[stepIndex+1 : closeBracketIndex]) if err != nil { return "", nil, -1, fmt.Errorf("invalid step value: %v", err) } } } logger.Debug("step value is", zap.Any("step", step)) if start >= end { return "", nil, -1, fmt.Errorf("start value (%d) cannot be greater than or equal to the end value (%d)", start, end) } if step > end || step == 0 { return "", nil, -1, fmt.Errorf("step value (%d) cannot be greater than the end value (%d) and greater than 0", step, end) } // count the indices for this range for i := start; i <= end; i += step { // if there is duplicated index and reject if _, exists := seenIndices[i]; exists { return "", nil, -1, fmt.Errorf("duplicate index detected: %d", i) } seenIndices[i] = struct{}{} indices = append(indices, i) } } else { // Handle single indices like 1, 2, 3 if part != "" { index, err := strconv.Atoi(part) if err != nil || index < 1 { return "", nil, -1, fmt.Errorf("index must be a positive integer") } if err != nil { return "", nil, -1, fmt.Errorf("invalid index value: %s", part) } if _, exists := seenIndices[index]; !exists { seenIndices[index] = struct{}{} indices = append(indices, index) } } } } } else if openBracketIndex != -1 || closeBracketIndex != -1 { return "", nil, -1, fmt.Errorf("invalid array job name '%s'", jobName) } else { // no "[]" /* if slotLimitIndex != -1 { arrayName = jobName[:slotLimitIndex] } else { arrayName = jobName }*/ arrayName = jobName } if slotLimit != "" && len(indices) > 0 { maxIndex := 0 for i, index := range indices { if i == 0 || index > maxIndex { maxIndex = index } } if slotLimitValue > maxIndex { return "", nil, -1, fmt.Errorf("slotLimit (%d) cannot be greater than the maximum index (%d)", slotLimitValue, maxIndex) } } logger.Debug("Finish createArrayName function", zap.Any("arrayName", arrayName), zap.Any("indices", indices), zap.Any("slotLimitValue", slotLimitValue)) return arrayName, indices, slotLimitValue, nil } func isValidArrayName(arrayName string) error { // define the validate chars if len(arrayName) > 4094 { return fmt.Errorf("job name length exceeds the limit of 4094 characters") } validArrayName := regexp.MustCompile(`^[a-zA-Z0-9._-]+$`) if !validArrayName.MatchString(arrayName) { return fmt.Errorf("arrayName '%s' contains invalid characters. Allowed characters are [a-zA-Z0-9._-]", arrayName) } return nil } func isValidJobId(jobID string) error { matched, _ := regexp.MatchString(`^\d+$`, jobID) matchedIndex, _ := regexp.MatchString(`^\d+\[\d+\]$`, jobID) if !matched && !matchedIndex { return fmt.Errorf("jobid '%s' contains invalid characters. Allowed characters are 0-9 or with index value", jobID) } return nil } func validateAskedHosts(askedHosts string) error { if askedHosts == "" { return nil // Nothing to validate } // Split the AskedHosts string by spaces hostList := strings.Fields(askedHosts) if len(hostList) == 0 { // Handle case where AskedHosts might be just whitespace logger.Warn("AskedHosts field contains only whitespace.", zap.String("AskedHosts", askedHosts)) return nil } /* Validate each hostname against the cache for _, host := range hostList { if _, ok := store.Hosts.Get(host); !ok { err := fmt.Errorf("invalid hostname specified in AskedHosts: '%s'", host) logger.Error(err.Error()) return err // Return the specific error } }*/ logger.Debug("AskedHosts validated successfully.", zap.Strings("hosts", hostList)) return nil // All hosts are valid } // handle TransferFiles field func parseTransferFiles(transferFileStr string) (localFile, operator, remoteFile string, err error) { // remove space char remoteFile = "" transferFileStr = strings.TrimSpace(transferFileStr) parts := strings.Fields(transferFileStr) if len(parts) < 2 { return "", "", "", errors.New("TransferFiles format is invalid, expected format: local_file operator [remote_file]") } localFile = parts[0] operator = parts[1] if parts[0] == "<" || parts[0] == "<<" { operator = parts[0] pathParts := strings.Split(parts[1], "/") localFile = pathParts[len(pathParts)-1] remoteFile = parts[1] } if len(parts) > 2 { remoteFile = parts[2] } // validate the operator validOperators := []string{">", "<", "<<", "<>", "><"} if !utils.ContainsField(validOperators, operator) { return "", "", "", fmt.Errorf("invalid operator: %s, supported operators are: %v", operator, validOperators) } return localFile, operator, remoteFile, nil } func processJobRestartRequest(newJob *models.JobInfo, req *models.JobRestartRequest) error { logger.Debug("Enter processJobRestartRequest function", zap.Any("req", req)) if req.AskedHosts != "" { newJob.AskedHosts = req.AskedHosts } return nil } func processJobRequest(ctx *gin.Context, req *models.JobSubmitRequest, defaultValues map[string]string) error { if req.InFile != "" { req.InFile = strings.ReplaceAll(req.InFile, "%J", defaultValues["job_id"]) } if req.OutFile != "" { req.OutFile = strings.ReplaceAll(req.OutFile, "%J", defaultValues["job_id"]) } if req.ErrFile != "" { req.ErrFile = strings.ReplaceAll(req.ErrFile, "%J", defaultValues["job_id"]) } if req.OutDir != "" { if defaultValues["job_group"] != "" { req.OutDir = strings.ReplaceAll(req.OutDir, "%JG", defaultValues["job_group"]) } req.OutDir = strings.ReplaceAll(req.OutDir, "%J", defaultValues["job_id"]) req.OutDir = strings.ReplaceAll(req.OutDir, "%U", defaultValues["user_name"]) req.OutDir = strings.ReplaceAll(req.OutDir, "%P", defaultValues["project_name"]) } if req.CWD != "" { req.CWD = strings.ReplaceAll(req.CWD, "%EJ", defaultValues["job_id"]) req.CWD = strings.ReplaceAll(req.CWD, "%J", defaultValues["job_id"]) if defaultValues["job_group"] != "" { req.CWD = strings.ReplaceAll(req.CWD, "%JG", defaultValues["job_group"]) } req.CWD = strings.ReplaceAll(req.CWD, "%P", defaultValues["project_name"]) req.CWD = strings.ReplaceAll(req.CWD, "%U", defaultValues["user_name"]) } // initialize extendDataMap extendDataMap := make(map[string]interface{}) if req.ExtendData != "" { _ = json.Unmarshal([]byte(req.ExtendData), &extendDataMap) } if req.TransferFiles != nil { // Copy local files and check for errors result, err := copyLocalFilesConcurrently(ctx, realCopyLocalFiles, defaultValues["user_name"]) if err != nil { return fmt.Errorf("failed to process local files: %v", err) } // prepare // Update ExtendData with copied URI var extendData []map[string]interface{} if err := json.Unmarshal([]byte(req.ExtendData), &extendData); err != nil { return fmt.Errorf("failed to parse ExtendData JSON: %v", err) } if len(result) != 0 { for i, entry := range extendData { if localFile, ok := entry["local_file"].(string); ok { if copiedURI, exists := result[localFile]; exists { entry["local_file"] = copiedURI } } extendData[i] = entry } } logger.Debug("In processJobRequest function, after CopyLocalFilesConcurrently, result is:", zap.Any("result", result)) // copy remote files locally and append to result remoteResult := copyRemoteFilesLocally(ctx, fakeCopyRemoteFile, defaultValues["user_name"]) for remoteFile, copiedURI := range remoteResult { result[remoteFile] = copiedURI } if len(remoteResult) != 0 { for i, entry := range extendData { if localFile, ok := entry["remote_file"].(string); ok { if copiedURI, exists := result[localFile]; exists { entry["remote_file"] = copiedURI } } extendData[i] = entry } } extendDataMap["transfer_files"] = extendData } // handle env into json format if req.Env != nil { envJsonb, err := utils.ConvertStringToJsonb(req.Env) if err != nil { return fmt.Errorf("Failed to convert env to JSON: %v", err) } var envData interface{} err = json.Unmarshal([]byte(envJsonb.String), &envData) if err != nil { logger.Error("Failed to unmarshal envJsonb string", zap.Error(err), zap.String("envJsonb", envJsonb.String)) return fmt.Errorf("failed to process environment variables: %v", err) } extendDataMap["sub_env_vars"] = envData } // handle limits limits := make(map[string]interface{}) if req.CPULimit.Seconds != 0 { limits["cpu_limit"] = req.CPULimit.Seconds } if req.FileLimit.MB != 0 { limits["fsize_limit"] = req.FileLimit.MB } if req.DataLimit.MB != 0 { limits["data_limit"] = req.DataLimit.MB } if req.StackLimit.MB != 0 { limits["stack_limit"] = req.StackLimit.MB } if req.CoreLimit.MB != 0 { limits["core_limit"] = req.CoreLimit.MB } if req.ProcessLimit.MB != 0 { limits["process_limit"] = req.ProcessLimit.MB } if req.ThreadLimit.MB != 0 { limits["thread_limit"] = req.ThreadLimit.MB } if req.MemLimit.MB != 0 { limits["rss_limit"] = req.MemLimit.MB } if req.SwapLimit.MB != 0 { limits["swap_limit"] = req.SwapLimit.MB } if req.NumFiles != 0 { limits["nofile_limit"] = req.NumFiles } if req.RunLimit.Seconds != 0 { limits["run_limit"] = req.RunLimit.Seconds } if len(limits) > 0 { extendDataMap["limits"] = limits } if req.Interactive != 0 { extendDataMap["interactive"] = req.Interactive } if len(extendDataMap) > 0 { updatedExtendDataBytes, err := json.Marshal(extendDataMap) if err != nil { return fmt.Errorf("Failed to marshal updated ExtendData: %v", err) } req.ExtendData = string(updatedExtendDataBytes) } return nil } func postJobRequest(job *models.JobInfo) error { if job.InFile != "" { job.InFile = strings.ReplaceAll(job.InFile, "%I", strconv.Itoa(job.ArrayIdx)) } if job.OutFile != "" { job.OutFile = strings.ReplaceAll(job.OutFile, "%I", strconv.Itoa(job.ArrayIdx)) } if job.ErrFile != "" { job.ErrFile = strings.ReplaceAll(job.ErrFile, "%I", strconv.Itoa(job.ArrayIdx)) } if job.OutDir != "" { job.OutDir = strings.ReplaceAll(job.OutDir, "%I", strconv.Itoa(job.ArrayIdx)) } if job.CWD != "" { job.CWD = strings.ReplaceAll(job.CWD, "%I", strconv.Itoa(job.ArrayIdx)) } if len(job.InFile) > 4094 || len(job.OutFile) > 4094 || len(job.ErrFile) > 4094 || len(job.CWD) > 4094 { return errors.New("file length exceeds the limit of 4094 characters") } if len(job.PreExecCmd) > 4094 || len(job.PostExecCmd) > 4094 { return errors.New("command path exceeds the limit of 4094 characters") } if len(job.ChkpntDir) > 4000 { return errors.New("checkpoint directory length exceeds the limit of 4000 characters") } return nil } // ControlJobByID godoc // @Summary Perform an action on a job // @Description Submit an action to be performed on a specific job identified by jobids // @Tags jobs // @Accept json // @Produce json // @Param jobids path string true "Job ID" // @Param action body JobSignalRequest true "Action to be performed" // @Success 202 {object} JobControlResponseSummary "Action successfully accepted" // @Failure 400 {object} ErrorResponse "Invalid request body or parameters" // @Failure 404 {object} ErrorResponse "Job not found" // @Failure 500 {object} ErrorResponse "Internal server error" // @Router /jobs/{jobids}/control [post] func (j JobController) ControlJobByID(ctx *gin.Context) { logger.Debug("Enter ControlJobByID() function from ", zap.String("client_ip", ctx.ClientIP())) // get submit user from token token := ctx.GetHeader("Authorization") submitter, err := GetUsernameFromToken(token) if err != nil { logger.Error("Get username from token error", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } // Get job IDs from URL parameter jobIDsStr := ctx.Param("jobids") var req models.JobControlJobIDRequest 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 } // Validate command convertedReq := models.JobControlRequest{ Action: req.Action, Signal: req.Signal, Comment: req.Comment, ForceKill: req.ForceKill, Method: req.Method, } err = preCheckJobControlRequest(&convertedReq) if err != nil { logger.Error("Failed to send job control request", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } signalval := convertedReq.Signal logger.Debug("Receive job control request:", zap.String("client_ip", ctx.ClientIP()), zap.String("jobIds", jobIDsStr), zap.Any("control request:", req)) //filters := make([]utils.FilterCondition, 0) fields := models.JOB_DEFAULT_COLUMN for i, field := range fields { fields[i] = utils.GetMappedField(field) } // add a default order: sort=job_id:asc sortOrders := make([]utils.SortOrder, 0) sortOrders = append(sortOrders, utils.SortOrder{ Column: "unique_job_id", // use unique_job_id as primary key Ascending: true, }) var acttype int switch convertedReq.Action { case "kill": acttype = models.JOB_KILL case "stop": acttype = models.JOB_STOP case "resume": acttype = models.JOB_RESUME case "checkpoint": acttype = models.JOB_CHKPNT default: logger.Debug("Unsupported action: " + req.Action) ReturnError(ctx, common.INVALID_PARAMS_CODE, "Unsupported action: "+req.Action, nil) return } //1. get user input job list and validate the job existed or not // Extract job IDs using the new function jobids := extractJobNames(jobIDsStr) for i := range jobids { jobids[i] = strings.TrimSpace(jobids[i]) } logger.Debug("The input jobids", zap.Any("jobids", jobids)) mergeJqueryReply := make([]utils.OrderMap, 0) // store all user's input jobid allFiltCdArrays := make([]utils.FilterConditionArray, 0) for _, jobid := range jobids { compfilters := make(utils.CompFilterCondition) filtCdArrays := make([]utils.FilterConditionArray, 0) subORFilters := make([]utils.FilterCondition, 0) var err error jobid = strings.TrimSpace(jobid) // parse the jobid to support array list format jobid, err := parseArrayName(jobid) if err != nil { logger.Error("job id format is invalid", zap.Error(err)) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } logger.Debug("GetJobByIDs after parseArrayName", zap.Any("jobid", jobid)) if !utils.IsNumeric(jobid.KeyID) { err := errors.New("bad jobid: " + jobid.KeyID) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } mainFiltCond := utils.FilterCondition{ Field: "job_id", Operator: "eq", Value: jobid.KeyID, } // treat every section as an OR condition if jobid.Indices != nil { // handle every section in Indices list. for _, jidxsec := range jobid.Indices { st, ed, err := parseArrayIndexSection(jidxsec.Index) if err != nil { err := errors.New("bad job array index: " + jidxsec.Index) logger.Error(err.Error()) ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } if jidxsec.Step == 1 { subORFilters = append(subORFilters, utils.FilterCondition{ Field: "array_idx", Operator: "between", Value: "(" + strconv.Itoa(st) + "," + strconv.Itoa(ed) + ")", }) } else { // handle step > 1 idxList := make([]string, 0) for idx := st; idx <= ed; idx += jidxsec.Step { idxList = append(idxList, strconv.Itoa(idx)) } subORFilters = append(subORFilters, utils.FilterCondition{ Field: "array_idx", Operator: "in", Value: "(" + strings.Join(idxList, ",") + ")", }) } } } filtCdArrays = append(filtCdArrays, utils.FilterConditionArray{ MainFiltCond: mainFiltCond, SubFiltConds: subORFilters, }) allFiltCdArrays = append(allFiltCdArrays, filtCdArrays...) compfilters["job_id"] = filtCdArrays var job models.JobInfo compfiltercd, err := utils.ConvertToNewFilterArray(compfilters, &job) if err != nil { logger.Error(err.Error()) // TODO: the msg used here is not proper ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) return } var bookmark string = "" var limit int = utils.PER_PAGE_LIMIT_MAX for { queryOpts, err := utils.BuildQueryOptions(nil, compfiltercd, sortOrders, limit, bookmark) if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } logger.Debug("the job query request:", zap.Any("jquery request arrayjob:", compfiltercd), zap.Any("sortOrders", sortOrders), zap.Any("limit", limit), zap.String("pagetoken", bookmark), zap.Any("fields", fields)) jqueryReply, page_token, err := services.QueryData(services.TableJobInfo, queryOpts, fields) if err != nil { logger.Error("Failed to query job info.", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } // Here, page_token is escaped and will not be recognized by dbapi, // for internal use, need add Unescape here. bookmark, _ = url.QueryUnescape(page_token) mergeJqueryReply = append(mergeJqueryReply, jqueryReply...) if bookmark == "" { break } } // end for fetch job loop } // end of for logger.Debug("job list got:", zap.Any("jobInfoList", mergeJqueryReply)) //2. validate job status and control action matching jobInfoList := mergeJqueryReply successJobs := []map[string]interface{}{} failedJobs := []map[string]interface{}{} // check if specific jobids are exist in jobInfoList. for _, ujobFilterCdArray := range allFiltCdArrays { var jobid int64 var arrayIdx int foundJobs := make([]utils.OrderMap, 0) jobid, _ = strconv.ParseInt(ujobFilterCdArray.MainFiltCond.Value, 10, 64) if len(ujobFilterCdArray.SubFiltConds) == 0 { // for non index section sepcified; arrayIdx = 0 fjobs, err := foundJob(jobid, arrayIdx, jobInfoList) if err != nil { failedJobs = append(failedJobs, map[string]interface{}{ "job_id": jobid, "array_idx": arrayIdx, "msg": "job not found", }) continue } foundJobs = append(foundJobs, fjobs...) } else { for _, subFiltCond := range ujobFilterCdArray.SubFiltConds { if subFiltCond.Operator == "between" { // (st, ed) value := strings.Trim(subFiltCond.Value, "()") values := strings.Split(value, ",") st, _ := strconv.Atoi(values[0]) ed, _ := strconv.Atoi(values[1]) for i := st; i <= ed; i++ { arrayIdx = i fjobs, err := foundJob(jobid, arrayIdx, jobInfoList) if err != nil { failedJobs = append(failedJobs, map[string]interface{}{ "job_id": jobid, "array_idx": arrayIdx, "msg": "job not found", }) } else { foundJobs = append(foundJobs, fjobs...) } } } else if subFiltCond.Operator == "in" { // (1,4,7,...) value := strings.Trim(subFiltCond.Value, "()") values := strings.Split(value, ",") for _, v := range values { arrayidx, _ := strconv.ParseInt(v, 10, 64) arrayIdx = int(arrayidx) fjobs, err := foundJob(jobid, arrayIdx, jobInfoList) if err != nil { failedJobs = append(failedJobs, map[string]interface{}{ "job_id": jobid, "array_idx": arrayIdx, "msg": "job not found", }) } else { foundJobs = append(foundJobs, fjobs...) } } } } } for _, jobInfo := range foundJobs { // check rbac if job_owner, exist := jobInfo.Data["submit_user"]; exist { jobPermission, err := GetRequireResourcePermission(submitter, "job:control") if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } if jobPermission.DomainType == models.DomainTypeSelf { if job_owner.(string) != submitter { logger.Error("Permission denied", zap.String("perm", "job:control"), zap.String("logonUser", submitter)) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } } } if job_queue, exist := jobInfo.Data["queue_name"]; exist { queuePermission, err := GetRequireResourcePermission(submitter, "queue:control") if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } var accessQueues []string if queuePermission.DomainType == models.DomainTypeAll { accessQueues = store.GetAllQueueNames() } if queuePermission.DomainType == models.DomainTypeQueue { accessQueues = queuePermission.DomainValues } // check job:control permission if err != nil || len(accessQueues) == 0 || !utils.ContainsField(accessQueues, job_queue.(string)) { ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } } // check job status to see if can do corresponding action validatedSuccessJobs, validatedFailedJobs := validateJobControl(jobInfo, convertedReq.Action, convertedReq.Signal) for _, successJob := range validatedSuccessJobs { successJobs = append(successJobs, map[string]interface{}{ "unique_job_id": successJob["unique_job_id"].(int64), "job_id": successJob["job_id"].(int64), "array_idx": successJob["array_idx"].(int), "msg": successJob["msg"].(string), }) } for _, failedJob := range validatedFailedJobs { failedJobs = append(failedJobs, map[string]interface{}{ "unique_job_id": failedJob["unique_job_id"].(int64), "job_id": failedJob["job_id"].(int64), "array_idx": failedJob["array_idx"].(int), "msg": failedJob["msg"].(string), }) } } if len(successJobs) > 0 { for _, successJob := range successJobs { signalInt, _ := strconv.Atoi(signalval) jobCtrlEvt := models.JobControlEvent{ UniqueJobID: successJob["unique_job_id"].(int64), Signal: signalInt, Comment: convertedReq.Comment, ForceKill: convertedReq.ForceKill, Method: convertedReq.Method, } jobCtrlEvtBytes, err := json.Marshal(jobCtrlEvt) if err != nil { logger.Error("encode job control event error", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, "encode job control event error"+": "+err.Error(), nil) return } err = backend.ControlJob(uint64(successJob["unique_job_id"].(int64)), acttype, string(jobCtrlEvtBytes)) if err != nil { logger.Error("send job control to backend error", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil) return } } } } // response to user //logger.Debug("job ctrl accept resps:", zap.Any("jobCtrlAcceptResps", jobCtrlAcceptResps)) //ReturnSuccess(ctx, common.ACCEPTED, "Job control accepted.", jobCtrlAcceptResps) response, successMsg := BuildBatchOperationResponse(convertedReq.Action, "Job", successJobs, failedJobs) overallStatus := response["status"].(string) logger.Info("Finished processing job control request", zap.String("action", convertedReq.Action), zap.String("overallStatus", overallStatus), zap.Int("succeededCount", len(successJobs)), zap.Int("failedCount", len(failedJobs)), zap.String("client_ip", ctx.ClientIP())) ReturnSuccess(ctx, common.ACCEPTED, successMsg, response) return } // ControlJob godoc // @Summary Perform an action on filtered out jobs // @Description Submit an action to be performed on specific jobs identified by filter in request body. // @Tags jobs // @Accept json // @Produce json // @Param jobids path string true "Job ID" // @Param action body JobControlRequest true "Action to be performed" // @Success 202 {object} JobControlResponseSummary "Action successfully accepted" // @Failure 400 {object} ErrorResponse "Invalid request body or parameters" // @Failure 404 {object} ErrorResponse "Job not found" // @Failure 500 {object} ErrorResponse "Internal server error" // @Router /jobs/control [post] func (j JobController) ControlJob(ctx *gin.Context) { logger.Debug("Enter ControlJob() function from ", zap.String("client_ip", ctx.ClientIP())) // Get filter in request body var req models.JobControlRequest // get submit user from token token := ctx.GetHeader("Authorization") submitter, err := GetUsernameFromToken(token) if err != nil { logger.Error("Get username from token error", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } 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 } err = preCheckJobControlRequest(&req) if err != nil { logger.Error("Failed to send job control request", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) return } signalval := req.Signal logger.Debug("Receive job control request:", zap.Any("control request:", req), zap.String("client_ip", ctx.ClientIP())) filters := make([]utils.FilterCondition, 0) fields := models.JOB_DEFAULT_COLUMN for i, field := range fields { fields[i] = utils.GetMappedField(field) } // // TODO: need consider if we need put all job control action together. // var acttype int switch req.Action { case "kill": acttype = models.JOB_KILL case "stop": acttype = models.JOB_STOP case "resume": acttype = models.JOB_RESUME case "checkpoint": acttype = models.JOB_CHKPNT /* -------- case "requeue": acttype = models.JOB_REQUEUE case "mig": acttype = models.JOB_MIG case "modify": acttype = models.JOB_MODIFY case "switch": acttype = models.JOB_SWITCH case "post": acttype = models.JOB_POST case "read": acttype = models.JOB_READ case "top": acttype = models.JOB_TOP case "bot": acttype = models.JOB_BOT */ default: logger.Debug("Unsupported action: " + req.Action) ReturnError(ctx, common.INVALID_PARAMS_CODE, "Unsupported action: "+req.Action, nil) return } //query job according query filter logger.Debug("request filters user input: ", zap.Any("filters", req.Filter)) // TODO: validate the req.Filter filters = getFilters(req, submitter) logger.Debug("final request filters: ", zap.Any("filters", filters)) //1. fetch jobs from db with req.Filter jobList, err := FetchJobList(filters) if err != nil { logger.Error("fetch job list error", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) return } if len(jobList) == 0 { logger.Debug("no job found") ReturnSuccess(ctx, common.SUCCESS_CODE, common.SUCCESS_MSG, map[string]interface{}{ "count": 0, "rows": []interface{}{}, "page_token": "", }) return } var jobInfoList []utils.OrderMap if !req.Filter.AllJobs { // will query job order by submittime, will get the last submitted job. if len(jobList) > 0 && jobList[len(jobList)-1].Data["array_idx"] == 0 { jobInfoList = jobList[len(jobList)-1:] } else { jobInfoList = jobList } } else { jobInfoList = jobList } logger.Debug("job list got:", zap.Any("jobInfoList", jobInfoList)) //2. go through job list and fill in action_control successJobs := []map[string]interface{}{} failedJobs := []map[string]interface{}{} for _, job := range jobInfoList { // check rbac begin if job_owner, exist := job.Data["submit_user"]; exist { jobPermission, err := GetRequireResourcePermission(submitter, "job:control") if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } if jobPermission.DomainType == models.DomainTypeSelf { if job_owner.(string) != submitter { logger.Error("Permission denied", zap.String("perm", "job:control"), zap.String("logonUser", submitter)) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } } } if job_queue, exist := job.Data["queue_name"]; exist { queuePermission, err := GetRequireResourcePermission(submitter, "queue:control") if err != nil { logger.Error(err.Error()) ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } var accessQueues []string if queuePermission.DomainType == models.DomainTypeAll { accessQueues = store.GetAllQueueNames() } if queuePermission.DomainType == models.DomainTypeQueue { accessQueues = queuePermission.DomainValues } // check job:control permission if err != nil || len(accessQueues) == 0 || !utils.ContainsField(accessQueues, job_queue.(string)) { ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } } // check job status to see if can do corresponding action validatedSuccessJobs, validatedFailedJobs := validateJobControl(job, req.Action, req.Signal) for _, successJob := range validatedSuccessJobs { successJobs = append(successJobs, map[string]interface{}{ "unique_job_id": successJob["unique_job_id"].(int64), "job_id": successJob["job_id"].(int64), "array_idx": successJob["array_idx"].(int), "msg": successJob["msg"].(string), }) } for _, failedJob := range validatedFailedJobs { failedJobs = append(failedJobs, map[string]interface{}{ "unique_job_id": failedJob["unique_job_id"].(int64), "job_id": failedJob["job_id"].(int64), "array_idx": failedJob["array_idx"].(int), "msg": failedJob["msg"].(string), }) } } if len(successJobs) > 0 { signalInt, _ := strconv.Atoi(signalval) for _, successJob := range successJobs { jobCtrlEvt := models.JobControlEvent{ UniqueJobID: successJob["unique_job_id"].(int64), Signal: signalInt, Comment: req.Comment, ForceKill: req.ForceKill, Method: req.Method, } jobCtrlEvtBytes, err := json.Marshal(jobCtrlEvt) if err != nil { logger.Error("encode job control event error", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, "encode job control event error:"+": "+err.Error(), nil) return } err = backend.ControlJob(uint64(successJob["unique_job_id"].(int64)), acttype, string(jobCtrlEvtBytes)) if err != nil { logger.Error("send job control to backend error", zap.Error(err)) ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil) return } } } // response to user //logger.Debug("job ctrl accept resps:", zap.Any("jobCtrlAcceptResps", jobCtrlAcceptResps)) //ReturnSuccess(ctx, common.ACCEPTED, "Job control accepted.", jobCtrlAcceptResps) response, successMsg := BuildBatchOperationResponse(req.Action, "Job", successJobs, failedJobs) overallStatus := response["status"].(string) logger.Info("Finished processing job control request", zap.String("action", req.Action), zap.String("overallStatus", overallStatus), zap.Int("succeededCount", len(successJobs)), zap.Int("failedCount", len(failedJobs)), zap.String("client_ip", ctx.ClientIP())) ReturnSuccess(ctx, common.ACCEPTED, successMsg, response) return } func preCheckJobControlRequest(req *models.JobControlRequest) error { logger.Debug("Enter preCheckJobControlRequest") if err := utils.ValidateControlAction(req.Action, models.SupportedJobActions); err != nil { return err } req.Action = strings.ToLower(strings.TrimSpace(req.Action)) // don't change the calling sequence in here, // for stop/resume, it doesn't user input signal, // but it will set default value later if (req.Action == "stop" || req.Action == "resume") && req.Signal != "" { return errors.New("signal must be empty for stop or resume actions") } // set default signal value if not provided if req.Signal == "" { if defaultSignal, ok := models.DefaultActionSignal[strings.ToLower(req.Action)]; ok { req.Signal = defaultSignal } } // transform signal to numeric value if it is a string if req.Signal != "" && !utils.IsNumeric(req.Signal) { if sigNum, ok := models.JobSignalMapTable[strings.ToUpper(req.Signal)]; ok { req.Signal = strconv.Itoa(sigNum) } else { return fmt.Errorf("unsupported signal: %s", req.Signal) } } if req.Action == "kill" && req.Signal != "" { found := false if utils.IsNumeric(req.Signal) { for _, v := range models.JobSignalMapTable { if strconv.Itoa(v) == req.Signal { found = true break } } if found == false { return fmt.Errorf("unsupported signal: %s", req.Signal) } } else { if _, ok := models.JobSignalMapTable[req.Signal]; !ok { return fmt.Errorf("unsupported signal: %s", req.Signal) } } } if req.Comment != "" { if len(req.Comment) > 512 { return errors.New("comment length exceeds the limit of 512 characters") } } if req.Action == "checkpoint" { if req.Method != "" { if len(req.Method) > 64 { return errors.New("method length exceeds the limit of 64 characters") } } else { req.Method = "CRIU" } } else { req.Method = "" } return nil } func getFilters(req models.JobControlRequest, submitter string) []utils.FilterCondition { filters := make([]utils.FilterCondition, 0) if req.Filter.Users == "" { filters = append(filters, utils.FilterCondition{ Field: "submit_user", Operator: "eq", Value: strings.TrimSpace(submitter), }) } else { if req.Filter.Users != "all" { filters = append(filters, utils.FilterCondition{ Field: "submit_user", Operator: "eq", Value: strings.TrimSpace(req.Filter.Users), // TODO: need consider usergroup. }) } } if req.Filter.AllJobs { filters = append(filters, utils.FilterCondition{ Field: "job_status", Operator: "notin", Value: "(DONE,EXIT)", }) } // TODO: need consider array job ????? if req.Filter.JobName != "" { filters = append(filters, utils.FilterCondition{ Field: "job_name", Operator: "eq", Value: strings.TrimSpace(req.Filter.JobName), }) } if req.Filter.JobGroupName != "" { filters = append(filters, utils.FilterCondition{ Field: "job_group", Operator: "eq", Value: strings.TrimSpace(req.Filter.JobGroupName), }) } if req.Filter.QueueName != "" { filters = append(filters, utils.FilterCondition{ Field: "queue_name", Operator: "eq", Value: strings.TrimSpace(req.Filter.QueueName), }) } // TODO: need handle exec host and hostgroup as job query. // if req.Filter.ExecHosts != "" { filters = append(filters, utils.FilterCondition{ Field: "exec_hosts", Operator: "eq", Value: strings.TrimSpace(req.Filter.ExecHosts), }) } if req.Filter.SLA != "" { filters = append(filters, utils.FilterCondition{ Field: "sla", Operator: "eq", Value: strings.TrimSpace(req.Filter.SLA), }) } return filters } /* validate if job status and job control action is match: * eg: stop a non run/pend/ssuspend job (psusp/done/exit/ususpend); * resume a non ususpend/psusp job(run/pend/done/exit/ssuspend); * kill a non active job ( done/exit) */ func validateJobControl(jobinfo utils.OrderMap, action string, signal string) (successJobs []map[string]interface{}, failedJobs []map[string]interface{}) { var actionType int switch action { case "stop": actionType = models.JOB_STOP case "resume": actionType = models.JOB_RESUME case "kill": actionType = models.JOB_KILL case "checkpoint": actionType = models.JOB_CHKPNT default: actionType = models.JOB_KILL } if rawStatus, exists := jobinfo.Data["job_status"]; exists { if jobStatus, ok := rawStatus.(int64); ok { db_job_status := int(jobStatus) job_status := models.JobStateStr(db_job_status) jobStatus = int64(models.JOBSTAT[job_status]) if utils.IsControlActionAllowed(actionType, jobStatus, models.JobActionControlMap) { if action == "kill" && signal != "" { if !utils.IsControlActionAllowed(actionType, jobStatus, models.JobKillSignalControlMap) { failedJobs = append(failedJobs, map[string]interface{}{ "unique_job_id": jobinfo.Data["unique_job_id"].(int64), "job_id": jobinfo.Data["job_id"].(int64), "array_idx": int(jobinfo.Data["array_idx"].(int64)), "msg": fmt.Sprintf("job cannot be or "), }) } else { successJobs = append(successJobs, map[string]interface{}{ "unique_job_id": jobinfo.Data["unique_job_id"].(int64), "job_id": jobinfo.Data["job_id"].(int64), "array_idx": int(jobinfo.Data["array_idx"].(int64)), "msg": fmt.Sprintf("job is being <%s>", action), }) } return successJobs, failedJobs } successJobs = append(successJobs, map[string]interface{}{ "unique_job_id": jobinfo.Data["unique_job_id"].(int64), "job_id": jobinfo.Data["job_id"].(int64), "array_idx": int(jobinfo.Data["array_idx"].(int64)), "msg": fmt.Sprintf("job is being <%s>", action), }) } else { logger.Error("Action not allowed for job", zap.Any("jobStatus", jobStatus)) failedJobs = append(failedJobs, map[string]interface{}{ "unique_job_id": jobinfo.Data["unique_job_id"].(int64), "job_id": jobinfo.Data["job_id"].(int64), "array_idx": int(jobinfo.Data["array_idx"].(int64)), "msg": fmt.Sprintf("job cannot be <%s>", action), }) } } else { logger.Error("job_status is not an int64", zap.Any("job_status", rawStatus)) failedJobs = append(failedJobs, map[string]interface{}{ "unique_job_id": jobinfo.Data["unique_job_id"].(int64), "job_id": jobinfo.Data["job_id"].(int64), "array_idx": int(jobinfo.Data["array_idx"].(int64)), "msg": fmt.Sprintf("Invalid job_status"), }) } } else { logger.Error("job_status not found in jobInfo", zap.Any("rawStatus", rawStatus)) failedJobs = append(failedJobs, map[string]interface{}{ "unique_job_id": jobinfo.Data["unique_job_id"].(int64), "job_id": jobinfo.Data["job_id"].(int64), "array_idx": int(jobinfo.Data["array_idx"].(int64)), "msg": fmt.Sprintf("job_status not found for job"), }) } return successJobs, failedJobs } /* check if the user specify job is found in job list * when array_idx = 0, may be an array job, but without index section specify, will only compare jobId */ func foundJob(jobid int64, array_idx int, jobinfos []utils.OrderMap) ([]utils.OrderMap, error) { jobs := make([]utils.OrderMap, 0) for _, jobinfo := range jobinfos { if array_idx > 0 { if jobinfo.Data["job_id"].(int64) == jobid && int(jobinfo.Data["array_idx"].(int64)) == array_idx { jobs = append(jobs, jobinfo) } } else { if jobinfo.Data["job_id"].(int64) == jobid { jobs = append(jobs, jobinfo) } } } if len(jobs) > 0 { return jobs, nil } else { return nil, errors.New("job not found") } } func processJobData(rows []utils.OrderMap, pagetoken string) utils.QueryResult { logger.Debug("Enter processJobData() function", zap.Any("rows", rows)) for _, row := range rows { if v, exist := row.Data["job_status"]; exist { if v != nil { jobStatus, _ := utils.IntOrInt64(v) row.Data["job_status"] = models.JobStateStr(jobStatus) } } // Update job_name if this is an array job child (including idx=0) if jobName, exist := row.Data["job_name"]; exist { // Use IsArrayJob to detect array children, which covers // the first child (idx=0) that the old array_idx > 0 check missed. var isArray bool if uid, ok := row.Data["unique_job_id"].(int64); ok { isArray = backend.IsArrayJob(uint64(uid)) } if isArray { if arrayIdx, ok := row.Data["array_idx"].(int); ok { row.Data["job_name"] = fmt.Sprintf("%s[%d]", jobName, arrayIdx) } else if arrayIdx64, ok := row.Data["array_idx"].(int64); ok { row.Data["job_name"] = fmt.Sprintf("%s[%d]", jobName, arrayIdx64) } } else { row.Data["job_name"] = jobName } } // Update launch_type if exist if val, ok := row.Data["launch_type"]; ok && val != nil { if int64Val, _ := val.(int64); ok { code := int(int64Val) if str, exists := models.GetLaunchTypeString(code); exists { row.Data["launch_type"] = str } else { row.Data["launch_type"] = "unknown" } } } // Process JSONB fields based on predefined list for _, jsonbField := range models.JSONBJFields { 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 job info JSONB field.", zap.Error(err)) } } } } // Convert database fields back to user-recognizable fields for userField, dbField := range models.JobFieldMappings { //if v, exist := row.Data[dbField]; exist && v != nil { if v, exist := row.Data[dbField]; exist { row.Data[userField] = v // Replace the value in row.Fields with userField for i, field := range row.Fields { if field == dbField { row.Fields[i] = userField break } } // Remove the user field from the data map delete(row.Data, dbField) } } } return utils.QueryResult{ Rows: rows, Count: len(rows), Pagetoken: pagetoken, } } func copyLocalFilesConcurrently(ctx *gin.Context, files []string, username string) (map[string]string, error) { logger.Debug("Enter copyLocalFilesConcurrently function files:", zap.Any("files", files)) result := make(map[string]string) webUrl := utils.GetServerHostPort(ctx) for _, localFile := range files { if strings.HasPrefix(localFile, webUrl) { localFile = strings.TrimPrefix(localFile, webUrl+"/") } // Extract the date from the localFile path parts := strings.Split(localFile, "/") if len(parts) < 2 { logger.Error("Invalid file path format", zap.String("file", localFile)) return nil, fmt.Errorf("invalid file path format: %s", localFile) } dateParts := strings.Split(parts[0], "_") if len(dateParts) < 1 { logger.Error("Invalid date format in file path", zap.String("file", localFile)) return nil, fmt.Errorf("invalid date format in file path: %s", localFile) } date := dateParts[0] // The first part is the date logger.Debug("Extracted date from file path", zap.String("date", date)) localFile = filepath.Join("/home/", username, "./apiserver_uploads", date, parts[0]) fmt.Println("localFile=", localFile) // Check if the file exists if _, err := os.Stat(localFile); os.IsNotExist(err) { logger.Error("File does not exist", zap.String("file", localFile)) return nil, fmt.Errorf("file does not exist: %s", localFile) } result[parts[1]] = localFile } return result, nil } func copyRemoteFilesLocally(ctx *gin.Context, files []string, username string) map[string]string { result := make(map[string]string) webUrl := utils.GetServerHostPort(ctx) submissionDate := time.Now().Format("20060102") for _, localFile := range files { // Generate MD5 hash for the file content hasher := md5.New() contentHash := hasher.Sum(nil) // Generate UUID based on the MD5 hash contentUUID := uuid.NewMD5(uuid.NameSpaceOID, contentHash[:]).String() copiedURI := webUrl + "/" + submissionDate + "_" + contentUUID + "/" + filepath.Base(localFile) result[localFile] = copiedURI } return result } // parse [1-3] to start = 1, end=3 func parseArrayIndexSection(idxsec string) (int, int, error) { parts := strings.Split(idxsec, "-") if len(parts) != 2 { start, err := strconv.Atoi(strings.TrimSpace(parts[0])) if err != nil { return 0, 0, err } return start, start, nil } start, err := strconv.Atoi(strings.TrimSpace(parts[0])) if err != nil { return 0, 0, err } end, err := strconv.Atoi(strings.TrimSpace(parts[1])) if err != nil { return 0, 0, err } if start >= end { return 0, 0, fmt.Errorf("start index should be bigger than end index") } else { return start, end, nil } } // fetch job list from job_info table according filter condition // need special handling for job_name=arrayjob[1-3] and exec_hosts cases. func FetchJobList(filters []utils.FilterCondition) ([]utils.OrderMap, error) { logger.Debug("Enter fetchJobList() function", zap.Any("filter", filters)) tablename := services.TableJobInfo var datatype models.JobInfo limit := utils.PER_PAGE_LIMIT_MAX fields := models.JOB_DEFAULT_COLUMN sortOrders := make([]utils.SortOrder, 0) jobLists := make([]utils.OrderMap, 0) compfilters := make(utils.CompFilterCondition) for i, filter := range filters { if filter.Field == "exec_hosts" { hosts := strings.Split(strings.TrimSpace(filters[i].Value), ",") execHosts, err := GetHostList(hosts) if err != nil { return nil, err } //change the filter to get unique_job_id list from db. subFilters := make([]utils.FilterCondition, 0) subFilters = append(subFilters, utils.FilterCondition{ Field: "host_name", Operator: "in", Value: "(" + strings.Join(execHosts, ",") + ")", }) uniqueJobIds, err := FetchJobIdList(subFilters) if err != nil { return nil, err } if len(uniqueJobIds) == 0 { return nil, nil } //Replace filter condition from "exec_hosts" to "unique_job_id". uniqueJobIdStrs := make([]string, len(uniqueJobIds)) for i, uniqueJobId := range uniqueJobIds { uniqueJobIdStrs[i] = strconv.FormatInt(uniqueJobId, 10) } filters[i].Field = "unique_job_id" filters[i].Operator = "in" filters[i].Value = "(" + strings.Join(uniqueJobIdStrs, ",") } // special handle job_name[index] cases, here index can be [1-3], [1,3,5],[1-3,5-7] situations if filters[i].Field == "job_name" { filtCdArrays := make([]utils.FilterConditionArray, 0) jobNames := extractJobNames(filters[i].Value) for i := range jobNames { jobNames[i] = strings.TrimSpace(jobNames[i]) } logger.Debug("After extractJobNames jobNames is", zap.Any("jobNames", jobNames)) if !(strings.Contains(filters[i].Value, "[") && strings.Contains(filters[i].Value, "]")) { continue } for _, jobname := range jobNames { subORFilters := make([]utils.FilterCondition, 0) jobnameArry, err := parseArrayName(jobname) logger.Debug("After parseArrayName jobname is ", zap.Any("jobname", jobnameArry)) if err != nil { return nil, errors.New("job name format is invalid") } if err = isValidArrayName(jobnameArry.KeyID); err != nil { err := errors.New("bad jobname: " + jobnameArry.KeyID) return nil, err } // in job control filter, operator is "eq" var operator = "eq" // treat every section as an OR condition mainFiltCond := utils.FilterCondition{ Field: filters[i].Field, Operator: operator, Value: jobnameArry.KeyID, } if jobnameArry.Indices != nil { // array jobid: job_id:in:(101[1-3, 5-6]) // handle every section in Indices list. for _, jidxsec := range jobnameArry.Indices { st, ed, err := parseArrayIndexSection(jidxsec.Index) if err != nil { err := errors.New("bad job array index: " + jidxsec.Index) return nil, err } if jidxsec.Step == 1 { subORFilters = append(subORFilters, utils.FilterCondition{ Field: "array_idx", Operator: "between", Value: "(" + strconv.Itoa(st) + "," + strconv.Itoa(ed) + ")", }) } else { // handle step > 1 idxList := make([]string, 0) for idx := st; idx <= ed; idx += jidxsec.Step { idxList = append(idxList, strconv.Itoa(idx)) } subORFilters = append(subORFilters, utils.FilterCondition{ Field: "array_idx", Operator: "in", Value: "(" + strings.Join(idxList, ",") + ")", }) } } } filtCdArrays = append(filtCdArrays, utils.FilterConditionArray{ MainFiltCond: mainFiltCond, SubFiltConds: subORFilters, }) } compfilters["job_name"] = filtCdArrays } } // remove filter with jobarray[] from filters, for filter in compfilters need special handling. if len(compfilters) > 0 { var filtersNew []utils.FilterCondition for _, filter := range filters { found := false for k, _ := range compfilters { if filter.Field == k { found = true break } } if found { continue } filtersNew = append(filtersNew, filter) } filters = filtersNew } var filtercds []utils.Condition var compfiltercd utils.CompCondition var err error if len(filters) > 0 { filtercds, err = utils.ConvertToNewFilter(filters, &datatype) if err != nil { logger.Error(err.Error()) return nil, err } } if len(compfilters) > 0 { compfiltercd, err = utils.ConvertToNewFilterArray(compfilters, &datatype) if err != nil { logger.Error(err.Error()) return nil, err } } sortOrders = append(sortOrders, utils.SortOrder{ Column: "unique_job_id", Ascending: true, }) pagetokenStr := "" for { queryOpts, err := utils.BuildQueryOptions(filtercds, compfiltercd, sortOrders, limit, pagetokenStr) if err != nil { return nil, err } queryReply, pagetoken, err := services.QueryData(tablename, queryOpts, fields) if err != nil { return nil, err } jobLists = append(jobLists, queryReply...) if pagetoken == "" { break } pagetokenStr = pagetoken } logger.Debug("In fetchJobList() function", zap.Any("joblist", jobLists)) return jobLists, nil } /* fetch job id list from job_exec_hosts table with filters * exec_hosts:in:(h1,h2...) and return unique_job_id list. */ func FetchJobIdList(filters []utils.FilterCondition) ([]int64, error) { logger.Debug("Enter fetchJobIdList() function", zap.Any("filter", filters)) tablename := services.TableJobExecHosts var datatype models.JobExecHosts limit := utils.PER_PAGE_LIMIT_MAX fields := []string{"insert_seq", "unique_job_id"} sortOrders := make([]utils.SortOrder, 0) jobLists := make([]utils.OrderMap, 0) jobIdList := make([]int64, 0) filtercds, err := utils.ConvertToNewFilter(filters, &datatype) if err != nil { logger.Error(err.Error()) return nil, err } sortOrders = append(sortOrders, utils.SortOrder{ Column: "insert_seq", Ascending: true, }) pagetokenStr := "" for { queryOpts, err := utils.BuildQueryOptions(filtercds, nil, sortOrders, limit, pagetokenStr) if err != nil { return nil, err } queryReply, pagetoken, err := services.QueryData(tablename, queryOpts, fields) if err != nil { return nil, err } jobLists = append(jobLists, queryReply...) if pagetoken == "" { break } pagetokenStr = pagetoken } if len(jobLists) > 0 { for _, job := range jobLists { jobIdList = append(jobIdList, job.Data["unique_job_id"].(int64)) } } // when a parrallel job runs on multiple host, different host will map to the same unique_job_id. // need remove duplicate unique_job_id. uniqJobIds := utils.RemoveDuplicates(jobIdList) logger.Debug("fetchJobIdList() function", zap.Any("jobIdList", uniqJobIds)) return uniqJobIds, nil } // user specified host/hostgroup list, get individual host list by query host/hostgroup cache; func GetHostList(hosts []string) ([]string, error) { execHosts := make([]string, 0) if len(hosts) == 0 { return nil, errors.New("exec_host is empty") } for _, host := range hosts { host = strings.TrimSpace(host) if host == "" { continue } // TODO: handle hostgroup expansion when hostgroup support is added. if _, ok := store.Hosts.Get(host); ok { execHosts = append(execHosts, host) } else { return nil, fmt.Errorf("invalid host/hostgroup %s", host) } } if len(execHosts) == 0 { return nil, errors.New("exec_host is empty") } return execHosts, nil }