lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/frontend/apiserver/controllers/jobs.go
Rongsong Shen 40917a01f0
feat(job-array-one-based-index): control by filter with element selections (task 7.5) 
Filter-based ControlJob validates a bracketed -J value before fetching
(400 with the parser's message; unbracketed names, including commands
with spaces, unchanged). FetchJobList evaluates bracketed job_name
entries through the selection predicates instead of unevaluated
compound conditions, so -J 'foo[1]' acts on element 1 of every array
named foo and never on ordinary jobs of that name. The most-recent-job
rule applies only without an element number and without all-jobs, and
treats the element of a one-element array like an ordinary job
(isSingleElementArray, a documented approximation). The old
index-section parser is deleted. Completes bug-1898 on the owned paths.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-10-01 12:40:09 +00:00

3225 lines
105 KiB
Go
Raw Blame History

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,
})
}
// ---- validate every filter entry before any store access ----
// (validate-all-first, job-array-one-based-index): the filter loop below
// can query the store (exec_hosts) and return early with count 0, so
// every entry -- including every job selection -- is checked here first
// and the first rejection, in request order, rejects the whole query.
// Entries with a bracketed job_id/job_name value become row predicates.
filters, selPreds, ok := validateJobQueryFilters(ctx, filters, typeValue)
if !ok {
return
}
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,
})
}
for i, filter := range filters {
// 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))
}
}
}
// ==================== 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 }
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
}
// ---- 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, nil, sortOrders, limit, pagetokenStr)
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil)
return
}
// job selections are evaluated as row predicates, before sorting and
// pagination (bug-1898)
queryOpts.Predicates = selPreds
logger.Debug("the job query request:", zap.Any("jquery request", filtercds), zap.Int("selection predicates", len(selPreds)), 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)
}
// validateJobQueryFilters is GetJobsInfo's validate-all-first pass. It
// checks every filter entry, in request order and before any store access:
// the filter field is supported for the query type, job_id/job_name values
// carry no index under type=array_stats, single-value operators have one
// value, and job selections parse and suit their operator
// (BuildSelectionPredicates). On success it returns the plain filters (the
// converted bracketed entries removed, written bare identity keys normalised
// to decimal) and the selection predicates; on the first rejection it writes
// the 400 reply and returns ok == false.
func validateJobQueryFilters(ctx *gin.Context, filters []utils.FilterCondition,
typeValue string) ([]utils.FilterCondition, []func(utils.OrderMap) bool, bool) {
supported := models.JOB_SUPPORT_FILTERS
if typeValue == "array_stats" {
supported = models.ARRAY_JOB_SUPPORT_FILTERS
}
plain := make([]utils.FilterCondition, 0, len(filters))
var preds []func(utils.OrderMap) bool
for _, filter := range filters {
if !utils.ContainsField(supported, 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 nil, nil, false
}
if filter.Field == "job_id" || filter.Field == "job_name" {
values := extractJobNames(filter.Value)
for i := range values {
values[i] = strings.TrimSpace(values[i])
}
// statistics rows are arrays, not elements: no index allowed
// (checked before the selection parsers)
if typeValue == "array_stats" && strings.ContainsAny(filter.Value, "[]") {
err := errors.New(filter.Field + " 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 nil, nil, false
}
// only "in", "between", "notin" take multiple values
if !(filter.Operator == "in" || filter.Operator == "between" || filter.Operator == "notin") && len(values) > 1 {
err := fmt.Errorf("operator doesn't allow to use multiple values: %v", values)
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return nil, nil, false
}
}
entryPlain, entryPreds, err := BuildSelectionPredicates([]utils.FilterCondition{filter})
if err != nil {
returnSelectionError(ctx, filter.Value, err)
return nil, nil, false
}
plain = append(plain, entryPlain...)
preds = append(preds, entryPreds...)
}
return plain, preds, true
}
// returnSelectionError writes the 400 reply of a rejected job selection or
// key. A selection-parser rejection (*SelError) passes its message through
// unchanged (it names the selection as written); any other rejection keeps
// today's "Invalid parameters: <reason>" text.
func returnSelectionError(ctx *gin.Context, raw string, err error) {
var se *SelError
if errors.As(err, &se) {
logger.Debug("rejected job selection", zap.String("selection", raw),
zap.String("rule", se.Rule), zap.String("reason", se.Msg))
ReturnError(ctx, common.INVALID_PARAMS_CODE, se.Msg, nil)
return
}
logger.Debug("rejected job selection", zap.String("selection", raw), zap.Error(err))
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
}
// 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))
// Validate every key before any store query (validate-all-first): the
// first rejected key, in request order, rejects the whole request.
sels := make([]Selection, 0, len(jobids))
for _, jobid := range jobids {
sel, se := ParseIdentitySelection(jobid)
if se != nil {
returnSelectionError(ctx, jobid, se)
return
}
sels = append(sels, sel)
}
mergeJqueryReply := make([]utils.OrderMap, 0)
var page_token = ""
for _, sel := range sels {
sortOrders := make([]utils.SortOrder, 0)
var job interface{}
if typeValue == "array_stats" {
job = &models.ArrayJobInfo{}
} else {
job = &models.JobInfo{}
}
filtercds, err := utils.ConvertToNewFilter([]utils.FilterCondition{{
Field: "job_id",
Operator: "eq",
Value: strconv.FormatUint(sel.Key, 10),
}}, job)
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return
}
// An element selection matches only that array's elements in its
// range (5.2, 5.11); statistics rows are whole arrays, so there the
// index part selects the array as before.
var preds []func(utils.OrderMap) bool
if sel.Form != SelBare && typeValue != "array_stats" {
preds = append(preds, identityMatcher(sel))
}
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(filtercds, nil, sortOrders, limit, bookmark)
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil)
return
}
queryOpts.Predicates = preds
logger.Debug("the job query request:", zap.Any("jquery request", filtercds), zap.String("selection", sel.Raw), 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
arraySize, isArray, err := preCheckJobRequest(&req)
if err != nil {
logger.Error("Failed to submit job request", zap.Error(err))
// An array index-expression error is the user's input: 400 with the
// message unchanged (2.1-2.8). Every other pre-check error stays 500.
var ae *ArrayExprError
if errors.As(err, &ae) {
ReturnError(ctx, common.INVALID_PARAMS_CODE, ae.Msg, nil)
return
}
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
}
}
// Allocate the ID: an array (one-element arrays included, 1.2, 7.8)
// through the array allocator, which returns its first element (index
// 1); an ordinary job through the plain allocator.
var jobID uint64
if isArray {
jobID, err = generateArrayJobID(clusterID, apiserverID, arraySize)
} else {
jobID, err = generateJobID(clusterID, apiserverID, 1)
}
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 is the first element for an array, so the submit-time
// %J/%I substitution gives <array-id>[1] and 1 (9.4).
primaryJobID := int64(jobID)
arrayJobID := int64(backend.ExtractArrayBaseID(jobID))
if isArray {
logger.Info("Generated job ID", zap.String("array_jobid", backend.FormatJobID(arrayJobID)),
zap.Int("array_size", arraySize))
} else {
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: arrayJobID,
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 = 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
}
if isArray {
// Report the array identity and its size (3.1, 3.2).
logger.Info("Job submission successfully from ", zap.String("client_ip", ctx.ClientIP()),
zap.String("array_jobid", backend.FormatJobID(arrayJobID)), zap.Int("array_size", arraySize))
response := models.JobSubmitResponse{
Message: fmt.Sprintf("Job <%d> is submitted to queue <%s> (%d elements).",
arrayJobID, req.QueueName, arraySize),
JobID: arrayJobID,
QueueName: req.QueueName,
ArraySize: arraySize,
}
ReturnSuccess(ctx, common.CREATED, common.SUCCESS_MSG, response)
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
}
// validateResreqSyntax is the resource-request syntax check used by
// preCheckJobRequest; a variable so tests can substitute it (bug-1744).
var validateResreqSyntax = backend.ValidateResreqSyntax
// Backend seams used by CreateJob, ControlJobByID and the filter-based
// ControlJob; variables (like validateResreqSyntax) so the non-cgo handler
// tests (jobs_handler_test.go) can replace them and observe ID allocation,
// submission and control dispatch. Production code never reassigns them.
var (
generateJobID = backend.GenerateJobID
generateArrayJobID = backend.GenerateArrayJobID
submitJob = services.SubmitJob
controlJob = backend.ControlJob
)
// preCheckJobRequest validates and normalizes a submit request before any ID
// is allocated. It returns the number of jobs to create and whether they form
// an array: (1, false, nil) for an ordinary job; (n, true, nil) for a
// bracketed name accepted by ParseSubmitIndexExpr (req.JobName is then the
// name without its index part); (0, false, err) on rejection, where an
// index-expression error is an unwrapped *ArrayExprError (CreateJob maps it
// to 400).
func preCheckJobRequest(req *models.JobSubmitRequest) (n int, isArray bool, err 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 0, false, 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 0, false, 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 0, false, fmt.Errorf("cannot submit job to queue %s: queue is closed", req.QueueName)
}
}
if req.JobName != "" {
// A name containing '[' or ']' is an array index expression; it is
// checked first so the interactive rule (2.1) is reached (2.9).
if strings.ContainsAny(req.JobName, "[]") {
arrayName, count, _, ae := ParseSubmitIndexExpr(req.JobName,
req.Interactive != 0, utils.JobYamlConfig.MaxArraySize)
if ae != nil {
logger.Error("job array index expression rejected",
zap.String("rule", ae.Rule), zap.String("reason", ae.Msg))
return 0, false, ae // unwrapped; never a typed-nil error
}
req.JobName = arrayName
logger.Debug("preCheckJobRequest parsed array job name",
zap.String("jobName", req.JobName), zap.Int("arraySize", count))
return count, true, nil
}
// Names without brackets keep the existing path (1.4).
arrayName, idx, slotLimit, err := createArrayName(req.JobName)
if err != nil {
logger.Error("job name format is invalid", zap.Error(err))
return 0, false, errors.New("job array list format is invalid: " + err.Error())
}
if arrayName != "" {
if err = isValidArrayName(arrayName); err != nil {
logger.Error(err.Error())
return 0, false, 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 1, false, 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 0, false, 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 0, false, err
}
if req.ProjectName != "" {
if len(req.ProjectName) > 511 {
return 0, false, errors.New("project name length exceeds the limit of 511 characters")
}
}
if req.Resreq != "" {
if len(req.Resreq) > 4094 {
return 0, false, errors.New("resreq length exceeds the limit of 4094 characters")
}
// bug-1744: reject a malformed rusage[] clause now, instead of
// accepting the job and letting the scheduler silently drop it.
if err := validateResreqSyntax(req.Resreq); err != nil {
return 0, false, err
}
}
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 0, false, 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 0, false, errors.New("environment variable should use key=value format")
}
key := parseEnvs[0]
if strings.TrimSpace(key) == "" {
return 0, false, 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 0, false, 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 0, false, 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 0, false, 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 0, false, 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 0, false, 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 0, false, 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 0, false, 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 0, false, errors.New("Launch_type must be openmpi or mpich")
}
req.LaunchType = launchType
}
return 1, false, 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. parse and validate every selection before any store query or
// dispatch (validate-all-first): the first rejected selection, in request
// order, rejects the whole command, which then acts on no job and reports
// no "not found" item.
jobids := extractJobNames(jobIDsStr)
for i := range jobids {
jobids[i] = strings.TrimSpace(jobids[i])
}
logger.Debug("The input jobids", zap.Any("jobids", jobids))
sels := make([]Selection, 0, len(jobids))
for _, jobid := range jobids {
sel, se := ParseIdentitySelection(jobid)
if se != nil {
returnSelectionError(ctx, jobid, se)
return
}
sels = append(sels, sel)
}
//2. match every selection against the store. The loop only collects:
// (a) the jobs to act on, de-duplicated by unique_job_id, in selection
// order, and (b) one "not found" item per missing number per selection
// (not de-duplicated). Nothing is validated or dispatched inside it
// (bug-1899: dispatching there re-sent earlier selections' jobs for every
// later selection).
actJobs := make([]utils.OrderMap, 0)
seenJobs := make(map[int64]bool)
notFoundJobs := []map[string]interface{}{}
for _, sel := range sels {
jobid := int64(sel.Key)
rows, ok := queryControlRowsByJobID(ctx, sel, fields, sortOrders)
if !ok {
return
}
foundJobs := make([]utils.OrderMap, 0)
if sel.Form == SelBare {
// no element part: every element of the array and any ordinary
// job with that identity (5.6)
fjobs, err := foundJob(jobid, 0, false, rows)
if err != nil {
notFoundJobs = append(notFoundJobs, map[string]interface{}{
"unique_job_id": jobid,
"job_id": jobid,
"array_idx": 0,
"msg": "job not found",
})
continue
}
foundJobs = append(foundJobs, fjobs...)
} else {
// explicit element numbers: only array elements with that
// number, never the whole array or an ordinary job (5.2, 5.4,
// 8.2); each missing number is reported with its would-be
// element ID (5.3, 5.4)
for k := sel.Lo; k <= sel.Hi; k++ {
fjobs, err := foundJob(jobid, k, true, rows)
if err != nil {
notFoundJobs = append(notFoundJobs, map[string]interface{}{
"unique_job_id": int64(backend.MakeElementID(sel.Key, k)),
"job_id": jobid,
"array_idx": k,
"msg": "job not found",
})
continue
}
foundJobs = append(foundJobs, fjobs...)
}
}
for _, jobInfo := range foundJobs {
uid, _ := jobInfo.Data["unique_job_id"].(int64)
if seenJobs[uid] {
continue
}
seenJobs[uid] = true
// 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 len(accessQueues) == 0 || !utils.ContainsField(accessQueues, job_queue.(string)) {
ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil)
return
}
}
actJobs = append(actJobs, jobInfo)
}
}
logger.Debug("jobs to act on:", zap.Any("jobs", actJobs))
//3. validate each job once against the action, then append the not-found
// items; every result item carries unique_job_id (5.1, 8.3).
successJobs := []map[string]interface{}{}
failedJobs := []map[string]interface{}{}
for _, jobInfo := range actJobs {
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),
})
}
}
failedJobs = append(failedJobs, notFoundJobs...)
//4. dispatch each accepted job once; a dispatch error is a 500.
signalInt, _ := strconv.Atoi(signalval)
for _, successJob := range successJobs {
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 = 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))
// A job name with '[' or ']' is an element selection: validate it before
// fetching (validate-all-first), so a rejection is a 400 with its own
// message. Names without brackets are matched as written, as before
// (unnamed jobs carry their command, with spaces, as their name).
hasElementNumber := strings.ContainsAny(req.Filter.JobName, "[]")
if hasElementNumber {
for _, filter := range filters {
if filter.Field != "job_name" {
continue
}
if _, _, err := BuildSelectionPredicates([]utils.FilterCondition{filter}); err != nil {
returnSelectionError(ctx, filter.Value, err)
return
}
}
}
//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 && !hasElementNumber {
// "most recent job" rule: rows are sorted by unique_job_id, so the
// last one is the most recent; act on it alone when it is an
// ordinary job (or a pre-upgrade element 0) or the only stored
// element of its array (5.10). A filter with an element number acts
// on exactly the selected elements (5.2).
last := jobList[len(jobList)-1]
if last.Data["array_idx"] == 0 || isSingleElementArray(last) {
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 = 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
}
// bug-1742: array_idx is a plain Go `int` on models.JobInfo, not `int64`
// -- utils.IntOrInt64 accepts either, unlike a bare `.(int64)` assertion.
arrayIdx, arrayOk := utils.IntOrInt64(jobinfo.Data["array_idx"])
if !arrayOk {
logger.Error("array_idx is not an int/int64", zap.Any("array_idx", jobinfo.Data["array_idx"]))
failedJobs = append(failedJobs, map[string]interface{}{
"unique_job_id": jobinfo.Data["unique_job_id"].(int64),
"job_id": jobinfo.Data["job_id"].(int64),
"array_idx": arrayIdx,
"msg": fmt.Sprintf("Invalid array_idx"),
})
return successJobs, failedJobs
}
if rawStatus, exists := jobinfo.Data["job_status"]; exists {
// bug-1742: job_status is likewise a plain Go `int`, not `int64`.
if jobStatusInt, ok := utils.IntOrInt64(rawStatus); ok {
jobStatus := int64(jobStatusInt)
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": arrayIdx,
"msg": fmt.Sprintf("job cannot be <killed> or <signaled>"),
})
} 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": arrayIdx,
"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": arrayIdx,
"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": arrayIdx,
"msg": fmt.Sprintf("job cannot be <%s>", action),
})
}
} else {
logger.Error("job_status is not an int/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": arrayIdx,
"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": arrayIdx,
"msg": fmt.Sprintf("job_status not found for job"),
})
}
return successJobs, failedJobs
}
// foundJob returns the rows of jobinfos a control selection names.
//
// With explicitIndex (the selection has an element part, "<id>[k]" or
// "<id>[a-b]"), only array elements -- rows whose unique_job_id carries the
// array marking -- of that array with array_idx == idx match, so "<id>[0]"
// never falls back to the whole array or to an ordinary job with identity
// <id> (5.4, 8.2). Without it, every row with that job_id matches: all
// elements of the array and any ordinary job (5.6).
func foundJob(jobID int64, idx int, explicitIndex bool, rows []utils.OrderMap) ([]utils.OrderMap, error) {
jobs := make([]utils.OrderMap, 0)
for _, row := range rows {
rowJobID, ok := rowUint64(row, "job_id")
if !ok || rowJobID != uint64(jobID) {
continue
}
if explicitIndex && !rowElementInRange(row, idx, idx) {
continue
}
jobs = append(jobs, row)
}
if len(jobs) == 0 {
return nil, errors.New("job not found")
}
return jobs, nil
}
// queryControlRowsByJobID fetches every stored row whose job_id is the
// selection's key (all pages), for ControlJobByID to match with foundJob. On
// a failure it writes the error reply and returns ok == false.
func queryControlRowsByJobID(ctx *gin.Context, sel Selection, fields []string,
sortOrders []utils.SortOrder) (rows []utils.OrderMap, ok bool) {
filtercds, err := utils.ConvertToNewFilter([]utils.FilterCondition{{
Field: "job_id",
Operator: "eq",
Value: strconv.FormatUint(sel.Key, 10),
}}, &models.JobInfo{})
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
return nil, false
}
rows = make([]utils.OrderMap, 0)
bookmark := ""
for {
queryOpts, err := utils.BuildQueryOptions(filtercds, nil, sortOrders, utils.PER_PAGE_LIMIT_MAX, bookmark)
if err != nil {
logger.Error(err.Error())
ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil)
return nil, false
}
logger.Debug("the job query request:", zap.String("selection", sel.Raw), zap.Any("filter", filtercds),
zap.String("pagetoken", bookmark), zap.Any("fields", fields))
reply, pageToken, 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 nil, false
}
rows = append(rows, reply...)
// page_token is escaped; unescape it for internal use
bookmark, _ = url.QueryUnescape(pageToken)
if bookmark == "" {
break
}
}
return rows, true
}
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
}
// fetch job list from job_info table according filter condition
// need special handling for job_name=arrayjob[1-3] (element selections,
// evaluated as row predicates) 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)
// Element selections (job_name values with '[' or ']') become row
// predicates; every other filter stays a plain condition. Parsed before
// the exec_hosts store lookups below (validate-all-first).
filters, preds, err := BuildSelectionPredicates(filters)
if err != nil {
return nil, err
}
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, ",")
}
}
var filtercds []utils.Condition
if len(filters) > 0 {
filtercds, err = utils.ConvertToNewFilter(filters, &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, nil, sortOrders, limit, pagetokenStr)
if err != nil {
return nil, err
}
queryOpts.Predicates = preds
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
}
Reference in a new issue View git blame Copy permalink