From c30e5f1d8256905e66e36435a1c747052f91ab08 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 1 Oct 2026 11:55:39 +0000 Subject: [PATCH] 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) --- .../specs/job-array-one-based-index/tasks.md | 2 +- src/frontend/apiserver/backend/handler.go | 2 +- .../apiserver/backend/handler_test.go | 29 ++ src/frontend/apiserver/controllers/jobs.go | 123 +++++--- .../controllers/jobs_handler_test.go | 263 ++++++++++++++++++ .../apiserver/controllers/jobs_test.go | 8 +- src/frontend/apiserver/models/jobs.go | 3 + 7 files changed, 390 insertions(+), 40 deletions(-) diff --git a/.kiro/specs/job-array-one-based-index/tasks.md b/.kiro/specs/job-array-one-based-index/tasks.md index 97422dfd..d7659332 100644 --- a/.kiro/specs/job-array-one-based-index/tasks.md +++ b/.kiro/specs/job-array-one-based-index/tasks.md @@ -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 diff --git a/src/frontend/apiserver/backend/handler.go b/src/frontend/apiserver/backend/handler.go index 935f153f..8b7577b8 100644 --- a/src/frontend/apiserver/backend/handler.go +++ b/src/frontend/apiserver/backend/handler.go @@ -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)) } diff --git a/src/frontend/apiserver/backend/handler_test.go b/src/frontend/apiserver/backend/handler_test.go index 0f7c93fb..de636691 100644 --- a/src/frontend/apiserver/backend/handler_test.go +++ b/src/frontend/apiserver/backend/handler_test.go @@ -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) + } +} diff --git a/src/frontend/apiserver/controllers/jobs.go b/src/frontend/apiserver/controllers/jobs.go index 7ae25885..dd303623 100644 --- a/src/frontend/apiserver/controllers/jobs.go +++ b/src/frontend/apiserver/controllers/jobs.go @@ -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 [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. diff --git a/src/frontend/apiserver/controllers/jobs_handler_test.go b/src/frontend/apiserver/controllers/jobs_handler_test.go index c9060bc0..afb165cc 100644 --- a/src/frontend/apiserver/controllers/jobs_handler_test.go +++ b/src/frontend/apiserver/controllers/jobs_handler_test.go @@ -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 (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 ." + 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 = [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 [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) + } + } +} diff --git a/src/frontend/apiserver/controllers/jobs_test.go b/src/frontend/apiserver/controllers/jobs_test.go index 52d7ee48..a7628fa9 100644 --- a/src/frontend/apiserver/controllers/jobs_test.go +++ b/src/frontend/apiserver/controllers/jobs_test.go @@ -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) } diff --git a/src/frontend/apiserver/models/jobs.go b/src/frontend/apiserver/models/jobs.go index 0ffdd63b..4dc0359a 100644 --- a/src/frontend/apiserver/models/jobs.go +++ b/src/frontend/apiserver/models/jobs.go @@ -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