feat(job-array-one-based-index): control by identity validates first and dispatches once (task 7.4)
ControlJobByID parses every selection with ParseIdentitySelection before any store access (first rejection: 400, nothing dispatched), collects the jobs to act on de-duplicated by unique_job_id plus one not-found result per missing number per selection (carrying the would-be element ID), validates each job once after the loop and dispatches each accepted job once. foundJob gains an explicit-index flag so <array-id>[0] matches only element 0 and never falls back to the whole array or an ordinary job. Fixes bug-1899 (earlier selections were re-dispatched for every later one). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
c3962e8f61
commit
4020740208
GPG key ID:
80533536B0EB651C
4 changed files with 475 additions and 241 deletions
|
|
@ -2147,207 +2147,85 @@ func (j JobController) ControlJobByID(ctx *gin.Context) {
|
|||
return
|
||||
}
|
||||
|
||||
//1. get user input job list and validate the job existed or not
|
||||
// Extract job IDs using the new function
|
||||
//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))
|
||||
mergeJqueryReply := make([]utils.OrderMap, 0)
|
||||
|
||||
// store all user's input jobid
|
||||
allFiltCdArrays := make([]utils.FilterConditionArray, 0)
|
||||
|
||||
sels := make([]Selection, 0, len(jobids))
|
||||
for _, jobid := range jobids {
|
||||
|
||||
compfilters := make(utils.CompFilterCondition)
|
||||
|
||||
filtCdArrays := make([]utils.FilterConditionArray, 0)
|
||||
subORFilters := make([]utils.FilterCondition, 0)
|
||||
|
||||
var err error
|
||||
|
||||
jobid = strings.TrimSpace(jobid)
|
||||
// parse the jobid to support array list format
|
||||
jobid, err := parseArrayName(jobid)
|
||||
if err != nil {
|
||||
logger.Error("job id format is invalid", zap.Error(err))
|
||||
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
|
||||
sel, se := ParseIdentitySelection(jobid)
|
||||
if se != nil {
|
||||
returnSelectionError(ctx, jobid, se)
|
||||
return
|
||||
}
|
||||
logger.Debug("GetJobByIDs after parseArrayName", zap.Any("jobid", jobid))
|
||||
sels = append(sels, sel)
|
||||
}
|
||||
|
||||
// gitea#45 follow-up: accept the human-readable job ID form
|
||||
// ("{cluster}-{apiserver}-{timestamp}-{seq}", e.g. what `jobs list`
|
||||
// prints) in addition to the numeric one xcecli already translates
|
||||
// to -- lets a caller hit this endpoint directly without
|
||||
// reimplementing xcecli's ID packing. Numeric IDs are unaffected.
|
||||
if !utils.IsNumeric(jobid.KeyID) {
|
||||
if packed, ok := backend.ParseHumanJobID(jobid.KeyID); ok {
|
||||
jobid.KeyID = strconv.FormatUint(packed, 10)
|
||||
} else {
|
||||
err := errors.New("bad jobid: " + jobid.KeyID)
|
||||
logger.Error(err.Error())
|
||||
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
|
||||
return
|
||||
}
|
||||
}
|
||||
//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{}{}
|
||||
|
||||
mainFiltCond := utils.FilterCondition{
|
||||
Field: "job_id",
|
||||
Operator: "eq",
|
||||
Value: jobid.KeyID,
|
||||
}
|
||||
|
||||
// treat every section as an OR condition
|
||||
if jobid.Indices != nil {
|
||||
// handle every section in Indices list.
|
||||
for _, jidxsec := range jobid.Indices {
|
||||
st, ed, err := parseArrayIndexSection(jidxsec.Index)
|
||||
if err != nil {
|
||||
err := errors.New("bad job array index: " + jidxsec.Index)
|
||||
logger.Error(err.Error())
|
||||
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
|
||||
return
|
||||
}
|
||||
|
||||
if jidxsec.Step == 1 {
|
||||
subORFilters = append(subORFilters, utils.FilterCondition{
|
||||
Field: "array_idx",
|
||||
Operator: "between",
|
||||
Value: "(" + strconv.Itoa(st) + "," + strconv.Itoa(ed) + ")",
|
||||
})
|
||||
} else { // handle step > 1
|
||||
idxList := make([]string, 0)
|
||||
for idx := st; idx <= ed; idx += jidxsec.Step {
|
||||
idxList = append(idxList, strconv.Itoa(idx))
|
||||
}
|
||||
subORFilters = append(subORFilters, utils.FilterCondition{
|
||||
Field: "array_idx",
|
||||
Operator: "in",
|
||||
Value: "(" + strings.Join(idxList, ",") + ")",
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
filtCdArrays = append(filtCdArrays, utils.FilterConditionArray{
|
||||
MainFiltCond: mainFiltCond,
|
||||
SubFiltConds: subORFilters,
|
||||
})
|
||||
|
||||
allFiltCdArrays = append(allFiltCdArrays, filtCdArrays...)
|
||||
|
||||
compfilters["job_id"] = filtCdArrays
|
||||
var job models.JobInfo
|
||||
compfiltercd, err := utils.ConvertToNewFilterArray(compfilters, &job)
|
||||
if err != nil {
|
||||
logger.Error(err.Error())
|
||||
// TODO: the msg used here is not proper
|
||||
ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil)
|
||||
for _, sel := range sels {
|
||||
jobid := int64(sel.Key)
|
||||
rows, ok := queryControlRowsByJobID(ctx, sel, fields, sortOrders)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
var bookmark string = ""
|
||||
var limit int = utils.PER_PAGE_LIMIT_MAX
|
||||
for {
|
||||
queryOpts, err := utils.BuildQueryOptions(nil, compfiltercd, sortOrders, limit, bookmark)
|
||||
if err != nil {
|
||||
logger.Error(err.Error())
|
||||
ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil)
|
||||
return
|
||||
}
|
||||
logger.Debug("the job query request:", zap.Any("jquery request arrayjob:", compfiltercd), zap.Any("sortOrders", sortOrders), zap.Any("limit", limit),
|
||||
zap.String("pagetoken", bookmark), zap.Any("fields", fields))
|
||||
jqueryReply, page_token, err := services.QueryData(services.TableJobInfo, queryOpts, fields)
|
||||
if err != nil {
|
||||
logger.Error("Failed to query job info.", zap.Error(err))
|
||||
ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil)
|
||||
return
|
||||
}
|
||||
// Here, page_token is escaped and will not be recognized by dbapi,
|
||||
// for internal use, need add Unescape here.
|
||||
bookmark, _ = url.QueryUnescape(page_token)
|
||||
mergeJqueryReply = append(mergeJqueryReply, jqueryReply...)
|
||||
if bookmark == "" {
|
||||
break
|
||||
}
|
||||
} // end for fetch job loop
|
||||
|
||||
} // end of for
|
||||
logger.Debug("job list got:", zap.Any("jobInfoList", mergeJqueryReply))
|
||||
|
||||
//2. validate job status and control action matching
|
||||
jobInfoList := mergeJqueryReply
|
||||
|
||||
successJobs := []map[string]interface{}{}
|
||||
failedJobs := []map[string]interface{}{}
|
||||
|
||||
// check if specific jobids are exist in jobInfoList.
|
||||
for _, ujobFilterCdArray := range allFiltCdArrays {
|
||||
var jobid int64
|
||||
var arrayIdx int
|
||||
foundJobs := make([]utils.OrderMap, 0)
|
||||
|
||||
jobid, _ = strconv.ParseInt(ujobFilterCdArray.MainFiltCond.Value, 10, 64)
|
||||
|
||||
if len(ujobFilterCdArray.SubFiltConds) == 0 { // for non index section sepcified;
|
||||
arrayIdx = 0
|
||||
fjobs, err := foundJob(jobid, arrayIdx, jobInfoList)
|
||||
if 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 {
|
||||
failedJobs = append(failedJobs, map[string]interface{}{
|
||||
"job_id": jobid,
|
||||
"array_idx": arrayIdx,
|
||||
"msg": "job not found",
|
||||
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 {
|
||||
for _, subFiltCond := range ujobFilterCdArray.SubFiltConds {
|
||||
if subFiltCond.Operator == "between" {
|
||||
// (st, ed)
|
||||
value := strings.Trim(subFiltCond.Value, "()")
|
||||
values := strings.Split(value, ",")
|
||||
st, _ := strconv.Atoi(values[0])
|
||||
ed, _ := strconv.Atoi(values[1])
|
||||
for i := st; i <= ed; i++ {
|
||||
arrayIdx = i
|
||||
fjobs, err := foundJob(jobid, arrayIdx, jobInfoList)
|
||||
if err != nil {
|
||||
failedJobs = append(failedJobs, map[string]interface{}{
|
||||
"job_id": jobid,
|
||||
"array_idx": arrayIdx,
|
||||
"msg": "job not found",
|
||||
})
|
||||
} else {
|
||||
foundJobs = append(foundJobs, fjobs...)
|
||||
}
|
||||
}
|
||||
} else if subFiltCond.Operator == "in" {
|
||||
// (1,4,7,...)
|
||||
value := strings.Trim(subFiltCond.Value, "()")
|
||||
values := strings.Split(value, ",")
|
||||
for _, v := range values {
|
||||
arrayidx, _ := strconv.ParseInt(v, 10, 64)
|
||||
arrayIdx = int(arrayidx)
|
||||
fjobs, err := foundJob(jobid, arrayIdx, jobInfoList)
|
||||
if err != nil {
|
||||
failedJobs = append(failedJobs, map[string]interface{}{
|
||||
"job_id": jobid,
|
||||
"array_idx": arrayIdx,
|
||||
"msg": "job not found",
|
||||
})
|
||||
} else {
|
||||
foundJobs = append(foundJobs, fjobs...)
|
||||
}
|
||||
}
|
||||
// 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")
|
||||
|
|
@ -2380,58 +2258,66 @@ func (j JobController) ControlJobByID(ctx *gin.Context) {
|
|||
accessQueues = queuePermission.DomainValues
|
||||
}
|
||||
// check job:control permission
|
||||
if err != nil || len(accessQueues) == 0 || !utils.ContainsField(accessQueues, job_queue.(string)) {
|
||||
if len(accessQueues) == 0 || !utils.ContainsField(accessQueues, job_queue.(string)) {
|
||||
ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil)
|
||||
return
|
||||
}
|
||||
}
|
||||
// check job status to see if can do corresponding action
|
||||
validatedSuccessJobs, validatedFailedJobs := validateJobControl(jobInfo, convertedReq.Action, convertedReq.Signal)
|
||||
actJobs = append(actJobs, jobInfo)
|
||||
}
|
||||
}
|
||||
logger.Debug("jobs to act on:", zap.Any("jobs", actJobs))
|
||||
|
||||
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),
|
||||
})
|
||||
}
|
||||
//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 _, 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),
|
||||
})
|
||||
}
|
||||
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),
|
||||
})
|
||||
}
|
||||
|
||||
if len(successJobs) > 0 {
|
||||
for _, successJob := range successJobs {
|
||||
signalInt, _ := strconv.Atoi(signalval)
|
||||
jobCtrlEvt := models.JobControlEvent{
|
||||
UniqueJobID: successJob["unique_job_id"].(int64),
|
||||
Signal: signalInt,
|
||||
Comment: convertedReq.Comment,
|
||||
ForceKill: convertedReq.ForceKill,
|
||||
Method: convertedReq.Method,
|
||||
}
|
||||
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...)
|
||||
|
||||
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
|
||||
}
|
||||
//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,
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -2922,31 +2808,73 @@ func validateJobControl(jobinfo utils.OrderMap, action string,
|
|||
return successJobs, failedJobs
|
||||
}
|
||||
|
||||
/* check if the user specify job is found in job list
|
||||
* when array_idx = 0, may be an array job, but without index section specify, will only compare jobId
|
||||
*/
|
||||
func foundJob(jobid int64, array_idx int, jobinfos []utils.OrderMap) ([]utils.OrderMap, error) {
|
||||
// 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 _, jobinfo := range jobinfos {
|
||||
if array_idx > 0 {
|
||||
// bug-1742: array_idx is a plain Go `int` on models.JobInfo, not
|
||||
// `int64` -- utils.IntOrInt64 accepts either.
|
||||
jobinfoArrayIdx, ok := utils.IntOrInt64(jobinfo.Data["array_idx"])
|
||||
if ok && jobinfo.Data["job_id"].(int64) == jobid &&
|
||||
jobinfoArrayIdx == array_idx {
|
||||
jobs = append(jobs, jobinfo)
|
||||
}
|
||||
} else {
|
||||
if jobinfo.Data["job_id"].(int64) == jobid {
|
||||
jobs = append(jobs, jobinfo)
|
||||
}
|
||||
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 jobs, nil
|
||||
} else {
|
||||
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 {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue