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:
parent
0318ac5f44
commit
c30e5f1d82
7 changed files with 390 additions and 40 deletions
|
|
@ -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))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue