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
|
|
@ -163,7 +163,7 @@
|
|||
- _Requirements: 5.2, 5.5, 5.6, 5.7, 5.8, 5.11, 5.12, 5.14, 5.16, 5.17_
|
||||
- _Depends: 5.4, 7.2_
|
||||
|
||||
- [ ] 7.4 Control jobs by identity: validate first, dispatch once (fixes bug-1899)
|
||||
- [x] 7.4 Control jobs by identity: validate first, dispatch once (fixes bug-1899)
|
||||
- Replace the index-section parsing in this handler with the REST selection parser; validate all selections before matching; give the element matcher an explicit-index flag so `<array-id>[0]` never falls back to the whole array or an ordinary job
|
||||
- Collect the de-duplicated jobs to act on and, separately, one not-found result per missing number per selection carrying the would-be element ID; validate each job once after the loop, then dispatch each accepted job once (a dispatch error stays a 500)
|
||||
- Update the existing element-matcher test to the new signature with a tagged fixture
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -976,3 +976,261 @@ func TestJobHandler_GetJobByIDsSelections(t *testing.T) {
|
|||
assert400(t, "5.14", f.getJobs("64-0-1-0"), "wider than allowed")
|
||||
assert400(t, "5.5", f.getJobs(X+"[3-1]"), "reversed")
|
||||
}
|
||||
|
||||
// ========== Control by identity (task 7.4, bug-1899) ==========
|
||||
|
||||
// controlItem is one accepted or failed item of a control reply.
|
||||
type controlItem struct {
|
||||
UniqueJobID uint64
|
||||
HasUnique bool
|
||||
JobID uint64
|
||||
ArrayIdx int
|
||||
Msg string
|
||||
}
|
||||
|
||||
// controlReply is the data of a 202 control reply.
|
||||
type controlReply struct {
|
||||
Status string
|
||||
Accepted []controlItem
|
||||
Failed []controlItem
|
||||
}
|
||||
|
||||
// decodeControlReply decodes a successful (202) control reply.
|
||||
func decodeControlReply(t *testing.T, w *httptest.ResponseRecorder) controlReply {
|
||||
t.Helper()
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200; body: %s", w.Code, w.Body.String())
|
||||
}
|
||||
dec := json.NewDecoder(bytes.NewReader(w.Body.Bytes()))
|
||||
dec.UseNumber()
|
||||
var raw struct {
|
||||
Code int `json:"code"`
|
||||
Data struct {
|
||||
Status string `json:"status"`
|
||||
Accepted []map[string]interface{} `json:"accepted"`
|
||||
Failed []map[string]interface{} `json:"failed"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := dec.Decode(&raw); err != nil {
|
||||
t.Fatalf("decode control reply: %v; body: %s", err, w.Body.String())
|
||||
}
|
||||
if raw.Code != 202 {
|
||||
t.Fatalf("code = %d, want 202; body: %s", raw.Code, w.Body.String())
|
||||
}
|
||||
num := func(m map[string]interface{}, k string) (uint64, bool) {
|
||||
n, ok := m[k].(json.Number)
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
v, err := strconv.ParseUint(n.String(), 10, 64)
|
||||
if err != nil {
|
||||
// job_id/unique_job_id are int64 on the wire; keep the bit pattern
|
||||
s, err2 := strconv.ParseInt(n.String(), 10, 64)
|
||||
if err2 != nil {
|
||||
t.Fatalf("%s %q: %v", k, n, err)
|
||||
}
|
||||
v = uint64(s)
|
||||
}
|
||||
return v, true
|
||||
}
|
||||
conv := func(in []map[string]interface{}) []controlItem {
|
||||
out := make([]controlItem, 0, len(in))
|
||||
for _, m := range in {
|
||||
var it controlItem
|
||||
it.UniqueJobID, it.HasUnique = num(m, "unique_job_id")
|
||||
it.JobID, _ = num(m, "job_id")
|
||||
idx, _ := num(m, "array_idx")
|
||||
it.ArrayIdx = int(idx)
|
||||
it.Msg, _ = m["msg"].(string)
|
||||
out = append(out, it)
|
||||
}
|
||||
return out
|
||||
}
|
||||
return controlReply{Status: raw.Data.Status, Accepted: conv(raw.Data.Accepted), Failed: conv(raw.Data.Failed)}
|
||||
}
|
||||
|
||||
// itemIDs returns the unique_job_id of every item, in reply order.
|
||||
func itemIDs(t *testing.T, items []controlItem) []uint64 {
|
||||
t.Helper()
|
||||
ids := make([]uint64, 0, len(items))
|
||||
for _, it := range items {
|
||||
if !it.HasUnique {
|
||||
t.Errorf("result item without unique_job_id: %+v", it)
|
||||
}
|
||||
ids = append(ids, it.UniqueJobID)
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
// assertExactIDs checks got equals want element by element (order kept).
|
||||
func assertExactIDs(t *testing.T, name string, got, want []uint64) {
|
||||
t.Helper()
|
||||
if len(got) != len(want) {
|
||||
t.Errorf("%s: %v, want %v", name, got, want)
|
||||
return
|
||||
}
|
||||
for i := range got {
|
||||
if got[i] != want[i] {
|
||||
t.Errorf("%s: %v, want %v", name, got, want)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// assertSeeded fails the test unless every id is in store.Jobs, so an
|
||||
// act-on-no-job assertion cannot pass on an empty store.
|
||||
func assertSeeded(t *testing.T, ids ...uint64) {
|
||||
t.Helper()
|
||||
for _, id := range ids {
|
||||
if !store.Jobs.Has(id) {
|
||||
t.Fatalf("fixture row %d missing from store.Jobs", id)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// controlByIDEscaped posts a control action for a selection list, escaping
|
||||
// the path segment.
|
||||
func (f *handlerFixture) controlByIDEscaped(jobids, action string) *httptest.ResponseRecorder {
|
||||
f.t.Helper()
|
||||
return f.controlByID(url.PathEscape(jobids), action)
|
||||
}
|
||||
|
||||
// TestJobHandler_ControlTwoSelectionsDispatchOnce (bug-1899): a request with
|
||||
// two selections dispatches each selected element exactly once and lists
|
||||
// each once; the same holds for two ordinary jobs.
|
||||
func TestJobHandler_ControlTwoSelectionsDispatchOnce(t *testing.T) {
|
||||
f := newHandlerFixture(t)
|
||||
X := handlerTestArrayBase + 30<<15
|
||||
Y := handlerTestArrayBase + 31<<15
|
||||
f.seedArrayElements(X, "foo", "RUN", 1, 2)
|
||||
f.seedArrayElements(Y, "bar", "RUN", 1, 2)
|
||||
A := handlerTestOrdinaryBase + 1
|
||||
B := handlerTestOrdinaryBase + 2
|
||||
f.seedOrdinaryJob(A, "a", "RUN")
|
||||
f.seedOrdinaryJob(B, "b", "RUN")
|
||||
|
||||
x1, y2 := backend.MakeElementID(X, 1), backend.MakeElementID(Y, 2)
|
||||
r := decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+"[1],"+jobIDString(Y)+"[2]", "kill"))
|
||||
assertExactIDs(t, "X[1],Y[2] dispatched", f.dispatchedIDs(), []uint64{x1, y2})
|
||||
assertExactIDs(t, "X[1],Y[2] accepted", itemIDs(t, r.Accepted), []uint64{x1, y2})
|
||||
if len(r.Failed) != 0 {
|
||||
t.Errorf("X[1],Y[2] failed items: %+v", r.Failed)
|
||||
}
|
||||
|
||||
f.dispatched = nil
|
||||
r = decodeControlReply(t, f.controlByIDEscaped(jobIDString(A)+","+jobIDString(B), "kill"))
|
||||
assertExactIDs(t, "A,B dispatched", f.dispatchedIDs(), []uint64{A, B})
|
||||
assertExactIDs(t, "A,B accepted", itemIDs(t, r.Accepted), []uint64{A, B})
|
||||
|
||||
// a job selected by two selections is acted on once (de-duplicated by
|
||||
// unique_job_id)
|
||||
f.dispatched = nil
|
||||
x2 := backend.MakeElementID(X, 2)
|
||||
r = decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+","+jobIDString(X)+"[2]", "kill"))
|
||||
assertExactIDs(t, "X,X[2] dispatched", f.dispatchedIDs(), []uint64{x1, x2})
|
||||
assertExactIDs(t, "X,X[2] accepted", itemIDs(t, r.Accepted), []uint64{x1, x2})
|
||||
}
|
||||
|
||||
// TestJobHandler_ControlRangeReportsMissing (5.3): X[1-3] with element 2
|
||||
// missing dispatches elements 1 and 3 and reports 2 as not found, carrying
|
||||
// its would-be element ID.
|
||||
func TestJobHandler_ControlRangeReportsMissing(t *testing.T) {
|
||||
f := newHandlerFixture(t)
|
||||
X := handlerTestArrayBase + 32<<15
|
||||
f.seedArrayElements(X, "foo", "RUN", 1, 3, 4)
|
||||
x1, x2, x3 := backend.MakeElementID(X, 1), backend.MakeElementID(X, 2), backend.MakeElementID(X, 3)
|
||||
|
||||
r := decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+"[1-3]", "kill"))
|
||||
assertExactIDs(t, "dispatched", f.dispatchedIDs(), []uint64{x1, x3})
|
||||
assertExactIDs(t, "accepted", itemIDs(t, r.Accepted), []uint64{x1, x3})
|
||||
assertExactIDs(t, "failed", itemIDs(t, r.Failed), []uint64{x2})
|
||||
if len(r.Failed) == 1 {
|
||||
it := r.Failed[0]
|
||||
if it.Msg != "job not found" || it.JobID != X || it.ArrayIdx != 2 {
|
||||
t.Errorf("not-found item = %+v, want job_id %d, array_idx 2, msg \"job not found\"", it, X)
|
||||
}
|
||||
}
|
||||
if r.Status != "partially_accepted" {
|
||||
t.Errorf("status = %q, want partially_accepted", r.Status)
|
||||
}
|
||||
|
||||
// a range on an identity no array has: every number not found, no job
|
||||
f.dispatched = nil
|
||||
Z := handlerTestArrayBase + 33<<15
|
||||
r = decodeControlReply(t, f.controlByIDEscaped(jobIDString(Z)+"[1-2]", "kill"))
|
||||
if got := f.dispatchedIDs(); len(got) != 0 {
|
||||
t.Errorf("Z[1-2] dispatched %v, want none", got)
|
||||
}
|
||||
assertExactIDs(t, "Z[1-2] failed", itemIDs(t, r.Failed),
|
||||
[]uint64{backend.MakeElementID(Z, 1), backend.MakeElementID(Z, 2)})
|
||||
}
|
||||
|
||||
// TestJobHandler_ControlElementZeroMissing (5.4): X[0] with no element 0
|
||||
// acts on no job -- not on the array, not on an ordinary job with identity
|
||||
// X -- and reports element 0 as not found.
|
||||
func TestJobHandler_ControlElementZeroMissing(t *testing.T) {
|
||||
f := newHandlerFixture(t)
|
||||
X := handlerTestArrayBase + 34<<15
|
||||
elems := f.seedArrayElements(X, "foo", "RUN", 1, 2)
|
||||
// an ordinary job sharing the number X (array_idx 0)
|
||||
f.seedOrdinaryJob(X, "ordinary", "RUN")
|
||||
assertSeeded(t, uint64(elems[0].Unique_JobID), uint64(elems[1].Unique_JobID), X)
|
||||
|
||||
for _, sel := range []string{"[0]", "[0-0]"} {
|
||||
f.dispatched = nil
|
||||
r := decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+sel, "kill"))
|
||||
if got := f.dispatchedIDs(); len(got) != 0 {
|
||||
t.Errorf("X%s dispatched %v, want none", sel, got)
|
||||
}
|
||||
if len(r.Accepted) != 0 {
|
||||
t.Errorf("X%s accepted %+v, want none", sel, r.Accepted)
|
||||
}
|
||||
assertExactIDs(t, "X"+sel+" failed", itemIDs(t, r.Failed), []uint64{backend.MakeElementID(X, 0)})
|
||||
if len(r.Failed) == 1 && r.Failed[0].Msg != "job not found" {
|
||||
t.Errorf("X%s failed msg = %q, want \"job not found\"", sel, r.Failed[0].Msg)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestJobHandler_ControlPreUpgradeElementZero (8.2, 8.3): a pre-upgrade
|
||||
// element 0 is dispatched alone, also when the same request selects another
|
||||
// element of that array, and is reported with its own unique_job_id.
|
||||
func TestJobHandler_ControlPreUpgradeElementZero(t *testing.T) {
|
||||
f := newHandlerFixture(t)
|
||||
X := handlerTestArrayBase + 35<<15
|
||||
f.seedArrayElements(X, "old", "RUN", 0, 1, 2)
|
||||
x0, x2 := backend.MakeElementID(X, 0), backend.MakeElementID(X, 2)
|
||||
|
||||
r := decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+"[0]", "kill"))
|
||||
assertExactIDs(t, "X[0] dispatched", f.dispatchedIDs(), []uint64{x0})
|
||||
assertExactIDs(t, "X[0] accepted", itemIDs(t, r.Accepted), []uint64{x0})
|
||||
|
||||
f.dispatched = nil
|
||||
r = decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+"[0],"+jobIDString(X)+"[2]", "kill"))
|
||||
assertExactIDs(t, "X[0],X[2] dispatched", f.dispatchedIDs(), []uint64{x0, x2})
|
||||
assertExactIDs(t, "X[0],X[2] accepted", itemIDs(t, r.Accepted), []uint64{x0, x2})
|
||||
}
|
||||
|
||||
// TestJobHandler_ControlValidThenInvalid (Terms): a valid selection followed
|
||||
// by an invalid one rejects the whole command with 400 and dispatches
|
||||
// nothing, without "not found" items.
|
||||
func TestJobHandler_ControlValidThenInvalid(t *testing.T) {
|
||||
f := newHandlerFixture(t)
|
||||
X := handlerTestArrayBase + 36<<15
|
||||
elems := f.seedArrayElements(X, "foo", "RUN", 1, 2)
|
||||
assertSeeded(t, uint64(elems[0].Unique_JobID))
|
||||
|
||||
cases := []struct{ sel, want string }{
|
||||
{jobIDString(X) + "[1]," + jobIDString(X) + "[40000]", "element numbers must be at most"},
|
||||
{jobIDString(X) + "[1]," + jobIDString(X) + "[3-1]", "reversed"},
|
||||
{jobIDString(X) + "[1]," + jobIDString(X) + "[a]", "invalid job identity or element selection"},
|
||||
{jobIDString(X) + "[1]," + jobIDString(uint64(elems[1].Unique_JobID)), "must be named as"},
|
||||
}
|
||||
for _, c := range cases {
|
||||
f.dispatched = nil
|
||||
assert400(t, c.sel, f.controlByIDEscaped(c.sel, "kill"), c.want)
|
||||
if got := f.dispatchedIDs(); len(got) != 0 {
|
||||
t.Errorf("%s: dispatched %v, want none", c.sel, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ package controllers
|
|||
|
||||
import (
|
||||
"errors"
|
||||
"apiserver/backend"
|
||||
"apiserver/models"
|
||||
"apiserver/store"
|
||||
"apiserver/utils"
|
||||
|
|
@ -179,13 +180,17 @@ func TestValidateJobControl_RejectsNonKillableStatus(t *testing.T) {
|
|||
func TestFoundJob_MatchesRealArrayIdxType(t *testing.T) {
|
||||
resetJobSubmitState()
|
||||
|
||||
// an array element: its unique_job_id carries the array marking
|
||||
const key = uint64(2)<<40 | uint64(1)<<20
|
||||
elem := backend.MakeElementID(key, 1)
|
||||
jobInfo := models.JobInfo{
|
||||
JobID: 1788186846,
|
||||
ArrayIdx: 1,
|
||||
Unique_JobID: int64(elem),
|
||||
JobID: int64(key),
|
||||
ArrayIdx: 1,
|
||||
}
|
||||
orderMap := store.StructToOrderMap(jobInfo)
|
||||
|
||||
found, err := foundJob(1788186846, 1, []utils.OrderMap{orderMap})
|
||||
found, err := foundJob(int64(key), 1, true, []utils.OrderMap{orderMap})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
|
|
@ -194,6 +199,49 @@ func TestFoundJob_MatchesRealArrayIdxType(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// TestFoundJob_ExplicitIndex (5.4, 5.6, 8.2): an explicit element number
|
||||
// matches only array elements with that number -- never an ordinary job with
|
||||
// array_idx 0 or the whole array -- while no element part matches every row
|
||||
// with that job_id.
|
||||
func TestFoundJob_ExplicitIndex(t *testing.T) {
|
||||
const key = uint64(2)<<40 | uint64(1)<<20
|
||||
row := func(uid uint64, idx int) utils.OrderMap {
|
||||
return store.StructToOrderMap(models.JobInfo{Unique_JobID: int64(uid), JobID: int64(key), ArrayIdx: idx})
|
||||
}
|
||||
ordinary := row(key, 0) // an ordinary job sharing the number
|
||||
e1 := row(backend.MakeElementID(key, 1), 1)
|
||||
e2 := row(backend.MakeElementID(key, 2), 2)
|
||||
withZero := []utils.OrderMap{ordinary, e1, e2, row(backend.MakeElementID(key, 0), 0)}
|
||||
noZero := []utils.OrderMap{ordinary, e1, e2}
|
||||
|
||||
uids := func(rows []utils.OrderMap) []int64 {
|
||||
out := []int64{}
|
||||
for _, r := range rows {
|
||||
out = append(out, r.Data["unique_job_id"].(int64))
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
if got, err := foundJob(int64(key), 0, true, noZero); err == nil {
|
||||
t.Errorf("[0] without element 0 matched %v, want not found", uids(got))
|
||||
}
|
||||
got, err := foundJob(int64(key), 0, true, withZero)
|
||||
if err != nil || len(got) != 1 || uint64(got[0].Data["unique_job_id"].(int64)) != backend.MakeElementID(key, 0) {
|
||||
t.Errorf("[0] with a pre-upgrade element 0 = %v, %v; want that element alone", uids(got), err)
|
||||
}
|
||||
got, err = foundJob(int64(key), 2, true, noZero)
|
||||
if err != nil || len(got) != 1 || uint64(got[0].Data["unique_job_id"].(int64)) != backend.MakeElementID(key, 2) {
|
||||
t.Errorf("[2] = %v, %v; want element 2 alone", uids(got), err)
|
||||
}
|
||||
if got, err := foundJob(int64(key), 3, true, noZero); err == nil {
|
||||
t.Errorf("[3] matched %v, want not found", uids(got))
|
||||
}
|
||||
got, err = foundJob(int64(key), 0, false, noZero)
|
||||
if err != nil || len(got) != 3 {
|
||||
t.Errorf("bare = %v, %v; want every row with that job_id", uids(got), err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestPreCheckJobRequest_RejectsMalformedResreq (bug-1744): a resource request
|
||||
// the syntax check rejects must fail the submission instead of being accepted
|
||||
// and later silently dropped by the scheduler.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue