feat(job-array-one-based-index): REST submit creates arrays numbered from 1 (task 6.2)

preCheckJobRequest returns (n, isArray, err): a bracketed job name is
parsed by ParseSubmitIndexExpr first, so name[1-N] and name[1] create
arrays (one-element arrays included) allocated with
generateArrayJobID, stored under the bare name; expression errors are
answered 400 with their specific message and create nothing; names
without brackets take the unchanged createArrayName path. The response
names the array (jobid = array ID, new array_size, "(n elements)"),
submit-time %I/%J use the first element, and the submit and dispatch
logs identify arrays by array ID. Fixes bug-1896 for REST and xcecli.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-01 11:55:39 +00:00
parent 0318ac5f44
commit c30e5f1d82
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
7 changed files with 390 additions and 40 deletions

View file

@ -128,7 +128,7 @@
- _Requirements: 1.5, 2.1, 2.2, 2.3, 2.4, 2.5, 2.6, 2.7, 2.8, 2.9_
- _Boundary: ApiSubmitExpr_
- [ ] 6.2 Create arrays from bracketed submissions and report the array (fixes bug-1896 for REST and the client)
- [x] 6.2 Create arrays from bracketed submissions and report the array (fixes bug-1896 for REST and the client)
- Make the job pre-check return the element count and whether the job is an array; call the expression parser only for bracketed names (passing the interactive flag and the configured limit), store the name without its index part, and keep the existing path for names without brackets
- Map expression errors to 400 with their message unchanged and keep every other pre-check error as today
- Allocate array IDs through the array wrapper (one-element arrays included); keep the first element for the submit-time `%I`/`%J` substitution

View file

@ -270,7 +270,7 @@ func dispatchJobArray(handler NotificationHandler, elements []*models.JobInfo, b
if nNew > 0 {
logger.Info("job array push received and expanded",
zap.Uint64("base_jobid", baseJobID),
zap.String("array_jobid", FormatJobID(int64(ExtractArrayBaseID(baseJobID)))),
zap.Uint64("array_size", arraySize),
zap.Int("new_records", nNew))
}

View file

@ -643,3 +643,32 @@ func TestDispatchConfigPush_EmptySlicesNoOp(t *testing.T) {
t.Errorf("Clusters store should be empty, got %d entries", n)
}
}
// TestDispatchJobArray_LogsArrayIdentity: the array push log names the
// array by its formatted array identity, not the raw first element ID (5.9).
func TestDispatchJobArray_LogsArrayIdentity(t *testing.T) {
resetStores()
logs := captureLog(t)
h := &DefaultHandler{}
key := uint64(3)<<40 | uint64(1)<<20 // array-capable key
first := MakeElementID(key, 1)
elements := []*models.JobInfo{
{Unique_JobID: int64(first), JobID: int64(key), ArrayIdx: 1},
{Unique_JobID: int64(first + 1), JobID: int64(key), ArrayIdx: 2},
}
if n := dispatchJobArray(h, elements, first, 2); n != 2 {
t.Fatalf("nNew = %d, want 2", n)
}
entries := logs.FilterMessage("job array push received and expanded").All()
if len(entries) != 1 {
t.Fatalf("got %d confirmation log lines, want 1", len(entries))
}
fields := entries[0].ContextMap()
if want := FormatJobID(int64(key)); fields["array_jobid"] != want {
t.Errorf("array_jobid = %v, want %q", fields["array_jobid"], want)
}
if _, ok := fields["base_jobid"]; ok {
t.Errorf("log still carries the raw base_jobid: %v", fields)
}
}

View file

@ -1391,9 +1391,16 @@ func (j JobController) CreateJob(ctx *gin.Context) {
}
// precheck the request parameter before generating the job ID
parsedIndices, err := preCheckJobRequest(&req)
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
}
@ -1416,22 +1423,31 @@ func (j JobController) CreateJob(ctx *gin.Context) {
}
}
// Determine array size: 1 for normal jobs, >1 for array jobs
arraySize := 1
if len(parsedIndices) > 1 {
arraySize = len(parsedIndices)
// 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)
}
// Generate job ID via C idgen (single call for both normal and array jobs)
jobID, err := generateJobID(clusterID, apiserverID, arraySize)
if err != nil {
logger.Error("Failed to generate job ID", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, "failed to generate job ID: "+err.Error(), nil)
return
}
// primaryJobID 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)
logger.Info("Generated job ID", zap.Int64("jobID", primaryJobID), zap.Int("arraySize", arraySize))
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),
@ -1453,7 +1469,7 @@ func (j JobController) CreateJob(ctx *gin.Context) {
// Build the single JobInfo to submit
job := models.JobInfo{
Unique_JobID: primaryJobID,
JobID: int64(backend.ExtractArrayBaseID(uint64(primaryJobID))),
JobID: arrayJobID,
ArrayIdx: backend.ExtractArrayIdx(uint64(primaryJobID)),
//SubmitTime: time.Now().Unix(), // backend will fill the submit time;
SubmitUser: submitUser,
@ -1512,6 +1528,21 @@ func (j JobController) CreateJob(ctx *gin.Context) {
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{
@ -1545,7 +1576,14 @@ var (
controlJob = backend.ControlJob
)
func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) {
// 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"
@ -1554,39 +1592,56 @@ func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) {
if req.QueueName == "lost_and_found" {
logger.Error("Job submission to reserved queue rejected",
zap.String("queue_name", req.QueueName))
return nil, fmt.Errorf(
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 nil, fmt.Errorf("invalid queue name: %s", 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 nil, fmt.Errorf("cannot submit job to queue %s: queue is closed", 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 nil, errors.New("job array list format is invalid: " + err.Error())
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 nil, errors.New("job array list format is invalid: " + 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 idx, nil
return 1, false, nil
} else {
if req.Command != "" {
if len(req.Command) > 4094 {
@ -1599,29 +1654,29 @@ func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) {
if req.JobDescription != "" {
if len(req.JobDescription) > 4094 {
return nil, errors.New("job description length exceeds the limit of 4094 characters")
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 nil, err
return 0, false, err
}
if req.ProjectName != "" {
if len(req.ProjectName) > 511 {
return nil, errors.New("project name length exceeds the limit of 511 characters")
return 0, false, errors.New("project name length exceeds the limit of 511 characters")
}
}
if req.Resreq != "" {
if len(req.Resreq) > 4094 {
return nil, errors.New("resreq length exceeds the limit of 4094 characters")
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 nil, err
return 0, false, err
}
}
@ -1662,16 +1717,16 @@ func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) {
// handle environment variables
if req.Env != nil {
if len(req.Env) > 4094 {
return nil, errors.New("environment variable length exceeds the limit of 4094 characters")
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 nil, errors.New("environment variable should use key=value format")
return 0, false, errors.New("environment variable should use key=value format")
}
key := parseEnvs[0]
if strings.TrimSpace(key) == "" {
return nil, errors.New("environment variable key cannot be empty")
return 0, false, errors.New("environment variable key cannot be empty")
}
}
}
@ -1684,7 +1739,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) {
for _, transferFileStr := range req.TransferFiles {
localFile, operator, remoteFile, err := parseTransferFiles(transferFileStr)
if err != nil {
return nil, err
return 0, false, err
}
/* based on different operator, check the local file and remote file
@ -1695,7 +1750,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) {
*/
if operator == ">" || operator == "<>" || operator == "><" {
if localFile == "" {
return nil, errors.New("local file is required for operator " + operator)
return 0, false, errors.New("local file is required for operator " + operator)
}
// Append to realCopyLocalFiles for later processing
realCopyLocalFiles = append(realCopyLocalFiles, localFile)
@ -1703,7 +1758,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) {
if operator == "<" || operator == "<<" {
if remoteFile == "" {
return nil, errors.New("remote file is required for operator " + operator)
return 0, false, errors.New("remote file is required for operator " + operator)
}
fakeCopyRemoteFile = append(fakeCopyRemoteFile, remoteFile)
}
@ -1718,7 +1773,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) {
// Convert inFileData to JSON and assign to req.InFile
extendDataJSON, err := json.Marshal(extendData)
if err != nil {
return nil, fmt.Errorf("failed to marshal InFile JSON: %v", err)
return 0, false, fmt.Errorf("failed to marshal InFile JSON: %v", err)
}
req.ExtendData = string(extendDataJSON)
}
@ -1730,7 +1785,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) {
minTasks, maxTasks, err := checkNumProcessors(minTasks, maxTasks)
if err != nil {
logger.Error("Failed to parse NumProcessors", zap.Any("numProcessors", req.NumProcessors), zap.Error(err))
return nil, fmt.Errorf("invalid NumProcessors specification: %w", err)
return 0, false, fmt.Errorf("invalid NumProcessors specification: %w", err)
}
req.NumProcessors = minTasks
req.MaxNumProcessors = maxTasks
@ -1739,22 +1794,22 @@ func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) {
}
if req.MigThreshold < 0 || req.MigThreshold > 2147483647 {
return nil, errors.New("Incorrect migration threshold value. Specify a number greater than or equal to zero.")
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 nil, errors.New("Interactive field must be 0, 1, or 2")
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 nil, errors.New("Launch_type must be openmpi or mpich")
return 0, false, errors.New("Launch_type must be openmpi or mpich")
}
req.LaunchType = launchType
}
return nil, nil
return 1, false, nil
}
// It returns minTasks, maxTasks, and an error if parsing fails or constraints are violated.

View file

@ -26,6 +26,9 @@ import (
"time"
"github.com/gin-gonic/gin"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"go.uber.org/zap/zaptest/observer"
)
// handlerTestUser is the user every handler fixture request is signed as.
@ -471,3 +474,263 @@ func TestJobHandler_SmokeControlErrorIs500(t *testing.T) {
t.Errorf("body does not carry the control error: %s", w.Body.String())
}
}
// ========== Array submission (task 6.2) ==========
// submitData decodes the JobSubmitResponse of a successful submit.
func submitData(t *testing.T, w *httptest.ResponseRecorder) models.JobSubmitResponse {
t.Helper()
if w.Code != http.StatusOK {
t.Fatalf("status = %d, want 200; body: %s", w.Code, w.Body.String())
}
resp := decodeHandlerResponse(t, w)
if resp.Code != 201 {
t.Fatalf("code = %d, want 201; body: %s", resp.Code, w.Body.String())
}
var data models.JobSubmitResponse
if err := json.Unmarshal(resp.Data, &data); err != nil {
t.Fatalf("decode data: %v", err)
}
return data
}
// assertNothingCreated checks no ID was allocated and nothing submitted (2.10).
func (f *handlerFixture) assertNothingCreated() {
f.t.Helper()
if len(f.generateCalls) != 0 || len(f.arrayCalls) != 0 {
f.t.Errorf("ID allocation called: generateJobID %v, generateArrayJobID %v; want none",
f.generateCalls, f.arrayCalls)
}
if subs := f.submissions(); len(subs) != 0 {
f.t.Errorf("submitJob called %d times, want 0", len(subs))
}
}
// TestJobHandler_SubmitArrayRange: arr[1-3] creates one 3-element array
// through generateArrayJobID, stores the name "arr" and reports the array
// identity and size (1.1, 1.3, 3.1, 3.2).
func TestJobHandler_SubmitArrayRange(t *testing.T) {
f := newHandlerFixture(t)
f.seedQueue("normal")
data := submitData(t, f.submit(map[string]interface{}{
"job_name": "arr[1-3]", "command": "sleep 1", "queue_name": "normal"}))
first := backend.MakeElementID(handlerTestArrayBase, 1)
if len(f.arrayCalls) != 1 || f.arrayCalls[0] != 3 {
t.Fatalf("generateArrayJobID calls = %v, want [3]", f.arrayCalls)
}
if len(f.generateCalls) != 0 {
t.Errorf("generateJobID calls = %v, want none", f.generateCalls)
}
subs := f.submissions()
if len(subs) != 1 {
t.Fatalf("submitJob calls = %d, want 1", len(subs))
}
s := subs[0]
if s.ArraySize != 3 {
t.Errorf("submitted array size = %d, want 3", s.ArraySize)
}
if s.Job.JobName != "arr" {
t.Errorf("stored job name = %q, want %q", s.Job.JobName, "arr")
}
if uint64(s.Job.Unique_JobID) != first || uint64(s.Job.JobID) != handlerTestArrayBase || s.Job.ArrayIdx != 1 {
t.Errorf("submitted ids = {unique %#x, job %#x, idx %d}, want {%#x, %#x, 1}",
s.Job.Unique_JobID, s.Job.JobID, s.Job.ArrayIdx, first, handlerTestArrayBase)
}
// Each element gets a distinct identity: first + i (1.3).
for i := 0; i < s.ArraySize; i++ {
if backend.ExtractArrayIdx(first+uint64(i)) != i+1 ||
backend.ExtractArrayBaseID(first+uint64(i)) != handlerTestArrayBase {
t.Errorf("element %d identity %#x does not decode to [%d] of the array", i, first+uint64(i), i+1)
}
}
if uint64(data.JobID) != handlerTestArrayBase {
t.Errorf("response jobid = %d, want array id %d", data.JobID, handlerTestArrayBase)
}
if data.ArraySize != 3 {
t.Errorf("response array_size = %d, want 3", data.ArraySize)
}
wantMsg := "Job <" + jobIDString(handlerTestArrayBase) + "> is submitted to queue <normal> (3 elements)."
if data.Message != wantMsg {
t.Errorf("response message = %q, want %q", data.Message, wantMsg)
}
}
// TestJobHandler_SubmitOneElementArray: name[1] is a one-element array
// (1.2, 7.8) through generateArrayJobID with n = 1.
func TestJobHandler_SubmitOneElementArray(t *testing.T) {
f := newHandlerFixture(t)
f.seedQueue("normal")
data := submitData(t, f.submit(map[string]interface{}{
"job_name": "name[1]", "command": "sleep 1", "queue_name": "normal"}))
if len(f.arrayCalls) != 1 || f.arrayCalls[0] != 1 {
t.Fatalf("generateArrayJobID calls = %v, want [1]", f.arrayCalls)
}
if len(f.generateCalls) != 0 {
t.Errorf("generateJobID calls = %v, want none", f.generateCalls)
}
subs := f.submissions()
if len(subs) != 1 || subs[0].ArraySize != 1 || subs[0].Job.JobName != "name" || subs[0].Job.ArrayIdx != 1 {
t.Fatalf("submissions = %+v, want one {size 1, name \"name\", idx 1}", subs)
}
if data.ArraySize != 1 {
t.Errorf("response array_size = %d, want 1", data.ArraySize)
}
if uint64(data.JobID) != handlerTestArrayBase {
t.Errorf("response jobid = %d, want array id %d", data.JobID, handlerTestArrayBase)
}
}
// TestJobHandler_SubmitArrayExprRejected: index-expression errors are 400
// with the parser's message unchanged, and nothing is created (2.x, 2.10).
func TestJobHandler_SubmitArrayExprRejected(t *testing.T) {
cases := []struct {
name string
jobName string
interactive int
wantMsg string
}{
{"2.5 start 0", "arr[0-3]", 0, "job name 'arr[0-3]': element numbers start at 1"},
{"2.1 interactive", "arr[1-3]", 1, "job name 'arr[1-3]': interactive jobs cannot be arrays"},
{"2.3 list", "arr[1,2]", 0, ""},
{"2.7 start above 1", "arr[5]", 0, ""},
{"2.8 above limit", "arr[1-1001]", 0, ""},
{"2.2 malformed", "arr]", 0, ""},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
f := newHandlerFixture(t)
f.seedQueue("normal")
want := tc.wantMsg
if want == "" {
_, _, _, ae := ParseSubmitIndexExpr(tc.jobName, tc.interactive != 0, 1000)
if ae == nil {
t.Fatalf("parser accepted %q", tc.jobName)
}
want = ae.Msg
}
w := f.submit(map[string]interface{}{"job_name": tc.jobName, "command": "sleep 1",
"queue_name": "normal", "interactive": tc.interactive})
if w.Code != http.StatusBadRequest {
t.Fatalf("status = %d, want 400; body: %s", w.Code, w.Body.String())
}
resp := decodeHandlerResponse(t, w)
if resp.Code != 400 || resp.Msg != want {
t.Errorf("response = {%d, %v}, want {400, %q}", resp.Code, resp.Msg, want)
}
f.assertNothingCreated()
})
}
}
// TestJobHandler_SubmitPlainNamesUnchanged: names without '[' or ']' keep
// today's errors and 500 status and never create an array (1.4).
func TestJobHandler_SubmitPlainNamesUnchanged(t *testing.T) {
cases := []struct{ jobName, wantMsg string }{
{"my job", "job array list format is invalid: arrayName 'my job' contains invalid characters. Allowed characters are [a-zA-Z0-9._-]"},
{"foo%5", "job array list format is invalid: arrayName 'foo%5' contains invalid characters. Allowed characters are [a-zA-Z0-9._-]"},
{"foo%", "job array list format is invalid: invalid slot limit value '': must be a positive integer"},
}
for _, tc := range cases {
t.Run(tc.jobName, func(t *testing.T) {
f := newHandlerFixture(t)
f.seedQueue("normal")
w := f.submit(map[string]interface{}{"job_name": tc.jobName, "command": "sleep 1", "queue_name": "normal"})
if w.Code != http.StatusInternalServerError {
t.Fatalf("status = %d, want 500; body: %s", w.Code, w.Body.String())
}
if resp := decodeHandlerResponse(t, w); resp.Code != 500 || resp.Msg != tc.wantMsg {
t.Errorf("response = {%d, %v}, want {500, %q}", resp.Code, resp.Msg, tc.wantMsg)
}
f.assertNothingCreated()
})
}
}
// TestJobHandler_SubmitOrdinaryUnchanged: an ordinary job is accepted (no
// typed-nil error leaks out of the pre-check), reported as before and its
// response carries no array_size (3.3).
func TestJobHandler_SubmitOrdinaryUnchanged(t *testing.T) {
f := newHandlerFixture(t)
f.seedQueue("normal")
w := f.submit(map[string]interface{}{"job_name": "plain", "command": "sleep 1", "queue_name": "normal"})
data := submitData(t, w)
if strings.Contains(w.Body.String(), "array_size") {
t.Errorf("ordinary submit response carries array_size: %s", w.Body.String())
}
wantMsg := "Job <" + jobIDString(handlerTestOrdinaryBase) + "> is submitted to default queue <normal>."
if data.Message != wantMsg || uint64(data.JobID) != handlerTestOrdinaryBase {
t.Errorf("response = {%d, %q}, want {%d, %q}", data.JobID, data.Message, handlerTestOrdinaryBase, wantMsg)
}
if len(f.generateCalls) != 1 || f.generateCalls[0] != 1 || len(f.arrayCalls) != 0 {
t.Errorf("allocation = generateJobID %v, generateArrayJobID %v; want [1], none", f.generateCalls, f.arrayCalls)
}
// Unnamed job: still ordinary.
w = f.submit(map[string]interface{}{"command": "sleep 2", "queue_name": "normal"})
submitData(t, w)
if len(f.arrayCalls) != 0 {
t.Errorf("unnamed job allocated an array: %v", f.arrayCalls)
}
}
// TestJobHandler_SubmitArrayFileNameSubstitution: at submission a new
// array's file names get the first element's values, %I = 1 and
// %J = <array-id>[1] (9.4).
func TestJobHandler_SubmitArrayFileNameSubstitution(t *testing.T) {
f := newHandlerFixture(t)
f.seedQueue("normal")
submitData(t, f.submit(map[string]interface{}{
"job_name": "arr[1-3]", "command": "sleep 1", "queue_name": "normal",
"out_file": "out.%J.%I", "err_file": "err.%J.%I"}))
subs := f.submissions()
if len(subs) != 1 {
t.Fatalf("submitJob calls = %d, want 1", len(subs))
}
// %J is the first element's identity, written <array-id>[1].
elem := backend.FormatJobID(int64(handlerTestArrayBase)) + "[1]"
if got := backend.FormatJobID(int64(backend.MakeElementID(handlerTestArrayBase, 1))); got != elem {
t.Fatalf("first element formats as %q, want %q", got, elem)
}
wantOut, wantErr := "out."+elem+".1", "err."+elem+".1"
if subs[0].Job.OutFile != wantOut || subs[0].Job.ErrFile != wantErr {
t.Errorf("files = {%q, %q}, want {%q, %q}", subs[0].Job.OutFile, subs[0].Job.ErrFile, wantOut, wantErr)
}
}
// TestJobHandler_SubmitArrayLogsArrayIdentity: CreateJob logs a new array
// by its array identity and size, not by the raw element ID (5.9).
func TestJobHandler_SubmitArrayLogsArrayIdentity(t *testing.T) {
f := newHandlerFixture(t)
f.seedQueue("normal")
core, logs := observer.New(zapcore.InfoLevel)
utils.Log = zap.New(core)
submitData(t, f.submit(map[string]interface{}{
"job_name": "arr[1-3]", "command": "sleep 1", "queue_name": "normal"}))
wantID := backend.FormatJobID(int64(handlerTestArrayBase))
for _, msg := range []string{"Generated job ID", "Job submission successfully from "} {
entries := logs.FilterMessage(msg).All()
if len(entries) != 1 {
t.Fatalf("log %q: %d entries, want 1", msg, len(entries))
}
fields := entries[0].ContextMap()
if fields["array_jobid"] != wantID {
t.Errorf("log %q: array_jobid = %v, want %q", msg, fields["array_jobid"], wantID)
}
if fields["array_size"] != int64(3) {
t.Errorf("log %q: array_size = %v (%T), want 3", msg, fields["array_size"], fields["array_size"])
}
if _, ok := fields["jobID"]; ok {
t.Errorf("log %q: carries the raw jobID for an array: %v", msg, fields)
}
}
}

View file

@ -25,7 +25,7 @@ func TestPreCheckJobRequest_LostAndFound(t *testing.T) {
QueueName: "lost_and_found",
}
_, err := preCheckJobRequest(req)
_, _, err := preCheckJobRequest(req)
if err == nil {
t.Fatal("expected error when submitting to lost_and_found queue, got nil")
}
@ -97,7 +97,7 @@ func TestPreCheckJobRequest_NormalQueue(t *testing.T) {
QueueName: "normal",
}
_, err := preCheckJobRequest(req)
_, _, err := preCheckJobRequest(req)
// Should not be rejected for the reserved-name check; may fail for other
// reasons (e.g. missing JobName validation), but not with our new error.
if err != nil && strings.Contains(err.Error(), "internal recovery queue") {
@ -115,7 +115,7 @@ func TestPreCheckJobRequest_EmptyQueue(t *testing.T) {
}
// Empty queue name sets default — should not trigger reserved-name error.
_, err := preCheckJobRequest(req)
_, _, err := preCheckJobRequest(req)
if err != nil && strings.Contains(err.Error(), "internal recovery queue") {
t.Errorf("empty queue should not trigger reserved-name error: %v", err)
}
@ -214,7 +214,7 @@ func TestPreCheckJobRequest_RejectsMalformedResreq(t *testing.T) {
req := &models.JobSubmitRequest{QueueName: "normal",
Resreq: "rusage[mem=100000000a0]"}
_, err := preCheckJobRequest(req)
_, _, err := preCheckJobRequest(req)
if checked != "rusage[mem=100000000a0]" {
t.Fatalf("resreq was not syntax-checked (checked %q)", checked)
}

View file

@ -710,6 +710,9 @@ type JobSubmitResponse struct {
JobID int64 `json:"jobid"`
QueueName string `json:"queue_name"`
Message string `json:"message"`
// ArraySize is the number of elements of a submitted array; omitted
// for an ordinary job (3.2, 3.3).
ArraySize int `json:"array_size,omitempty"`
}
// Job query struct in C lib