lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

submit job from xcecli

Browse source
This commit is contained in:
Rongsong Shen 2026-04-14 15:55:52 +08:00
parent c1c9f9e736
commit d9b5c361c2
5 changed files with 148 additions and 281 deletions
Download patch file Download diff file Expand all files Collapse all files

271
src/frontend/apiserver/controllers/jobs.go
View file

@ -27,7 +27,8 @@ import (
const (
//TODO: Need assign clusterID globally.
clusterID = 1
clusterID = 1
apiserverID = 0
)
var (
@ -36,10 +37,8 @@ var (
"resume": "CONT",
"stop": "STOP",
}
jobID_Worker = utils.NewWorker(clusterID)
realCopyLocalFiles []string
fakeCopyRemoteFile []string
indices []int
)
type JobController struct{}
@ -1183,16 +1182,17 @@ func (j JobController) RestartJobByID(ctx *gin.Context) {
}
}
jobIDArray, err := generateJobIDs(indices)
jobID, err := backend.GenerateJobID(clusterID, apiserverID, 1)
if err != nil {
ReturnError(ctx, common.ERROR_CODE, err.Error(), nil)
logger.Error("Failed to generate job ID for restart", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, "failed to generate job ID: "+err.Error(), nil)
return
}
// Use a concrete type for job
newJob := models.JobInfo{
Unique_JobID: jobIDArray[0].Unique_JobID,
JobID: jobIDArray[0].Unique_JobID,
Unique_JobID: int64(jobID),
JobID: int64(jobID),
ArrayIdx: 0,
SubmitUser: restarter,
JobStatus: models.JOBSTAT["PEND"],
@ -1341,10 +1341,9 @@ func (j JobController) RestartJobByID(ctx *gin.Context) {
if v, ok := oldJob["login_shell"].(string); ok {
newJob.LoginShell = v
}
logger.Info("Generated job IDs", zap.Any("jobIDArray", jobIDArray), zap.Any("indices", len(indices)))
logger.Info("Generated job ID for restart", zap.Int64("jobID", newJob.JobID))
// sequential job
err = services.SubmitJob(&newJob)
err = services.SubmitJob(&newJob, 1)
if err != nil {
logger.Error("Failed to submit job for restart process.", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil)
@ -1385,17 +1384,14 @@ func (j JobController) CreateJob(ctx *gin.Context) {
var req models.JobSubmitRequest
indices = []int{}
if err := utils.BindJSONStrict(ctx, &req); err != nil {
logger.Error("Invalid request parameters", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, err.Error(), nil)
return
}
// save the original job name before parsing
arrayJobName := req.JobName
// precheck the request parameter before generating the job ID
err = preCheckJobRequest(&req)
parsedIndices, err := preCheckJobRequest(&req)
if err != nil {
logger.Error("Failed to submit job request", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, err.Error(), nil)
@ -1419,20 +1415,23 @@ func (j JobController) CreateJob(ctx *gin.Context) {
return
}
}
// generate job IDs
jobIDArray, err := generateJobIDs(indices)
// Determine array size: 1 for normal jobs, >1 for array jobs
arraySize := 1
if len(parsedIndices) > 1 {
arraySize = len(parsedIndices)
}
// Generate job ID via C idgen (single call for both normal and array jobs)
jobID, err := backend.GenerateJobID(clusterID, apiserverID, arraySize)
if err != nil {
ReturnError(ctx, common.ERROR_CODE, err.Error(), 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
}
logger.Info("Generated job IDs", zap.Any("jobIDArray", jobIDArray), zap.Any("indices", len(indices)))
// get the primary jobID
var primaryJobID int64
if len(jobIDArray) > 0 {
primaryJobID = jobIDArray[0].Unique_JobID
}
primaryJobID := int64(jobID)
logger.Info("Generated job ID", zap.Int64("jobID", primaryJobID), zap.Int("arraySize", arraySize))
defaultValues := map[string]string{
"job_id": fmt.Sprintf("%d", primaryJobID),
@ -1451,96 +1450,68 @@ func (j JobController) CreateJob(ctx *gin.Context) {
return
}
jobs := make([]models.JobInfo, len(jobIDArray))
for i, jobID := range jobIDArray {
logger.Debug("Create new job info for jobID", zap.Int64("jobID", jobID.Unique_JobID))
// Create new job info
jobs[i] = models.JobInfo{
Unique_JobID: jobID.Unique_JobID,
JobID: primaryJobID,
ArrayIdx: jobID.ArrayIdx,
//SubmitTime: time.Now().Unix(), // backend will fill the submit time;
SubmitUser: submitUser,
JobStatus: models.JOBSTAT["PEND"], // new job is pend.
JobName: req.JobName,
Command: req.Command,
QueueName: req.QueueName, //TODO: need set to DEFAULT_QUEUE
ProjectName: req.ProjectName, //TODO: need set "default" if not specify?
UserGroup: req.UserGroup,
LicenseProject: req.LicenseProject,
JobDescription: req.JobDescription,
ResReq: req.Resreq,
RsvID: req.RsvID,
SLA: req.SLA,
MaxNumProcessors: req.MaxNumProcessors,
NumProcessors: req.NumProcessors,
AskedHosts: req.AskedHosts,
LaunchType: req.LaunchType,
//BeginTime: req.BeginTime,
//TermTime: req.TermTime,
BeginTime: models.CustomTime{Time: req.BeginTime.Time},
TermTime: models.CustomTime{Time: req.TermTime.Time},
JobPriority: req.JobPriority,
UserPriority: req.JobPriority,
DependCond: req.DependCond,
CWD: req.CWD,
FromHost: ctx.ClientIP(), //use ip now, need use hostname ?
PreExecCmd: req.PreExecCmd,
PostExecCmd: req.PostExecCmd,
RequeueErrorValues: req.RequeueErrorValues,
RunTimeEstimation: req.RunTimeEstimation,
MigThreshold: req.MigThreshold,
ChkpntPeriod: req.ChkpntPeriod,
ChkpntDir: req.ChkpntDir,
OutDir: req.OutDir,
InFile: req.InFile, // transfer file into json with absolute path
ErrFile: req.ErrFile,
OutFile: req.OutFile,
LoginShell: req.LoginShell,
ExtendData: models.Jsonb{String: req.ExtendData},
}
err = postJobRequest(&jobs[i])
if err != nil {
logger.Error("Failed to handle postJobRequest ", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, err.Error(), nil)
return
}
logger.Debug("jobinfo before submit.", zap.Any("job", jobs[i]))
// Build the single JobInfo to submit
job := models.JobInfo{
Unique_JobID: primaryJobID,
JobID: primaryJobID,
//SubmitTime: time.Now().Unix(), // backend will fill the submit time;
SubmitUser: submitUser,
JobStatus: models.JOBSTAT["PEND"], // new job is pend.
JobName: req.JobName,
Command: req.Command,
QueueName: req.QueueName, //TODO: need set to DEFAULT_QUEUE
ProjectName: req.ProjectName, //TODO: need set "default" if not specify?
UserGroup: req.UserGroup,
LicenseProject: req.LicenseProject,
JobDescription: req.JobDescription,
ResReq: req.Resreq,
RsvID: req.RsvID,
SLA: req.SLA,
MaxNumProcessors: req.MaxNumProcessors,
NumProcessors: req.NumProcessors,
AskedHosts: req.AskedHosts,
LaunchType: req.LaunchType,
//BeginTime: req.BeginTime,
//TermTime: req.TermTime,
BeginTime: models.CustomTime{Time: req.BeginTime.Time},
TermTime: models.CustomTime{Time: req.TermTime.Time},
JobPriority: req.JobPriority,
UserPriority: req.JobPriority,
DependCond: req.DependCond,
CWD: req.CWD,
FromHost: ctx.ClientIP(), //use ip now, need use hostname ?
PreExecCmd: req.PreExecCmd,
PostExecCmd: req.PostExecCmd,
RequeueErrorValues: req.RequeueErrorValues,
RunTimeEstimation: req.RunTimeEstimation,
MigThreshold: req.MigThreshold,
ChkpntPeriod: req.ChkpntPeriod,
ChkpntDir: req.ChkpntDir,
OutDir: req.OutDir,
InFile: req.InFile, // transfer file into json with absolute path
ErrFile: req.ErrFile,
OutFile: req.OutFile,
LoginShell: req.LoginShell,
ExtendData: models.Jsonb{String: req.ExtendData},
}
if len(jobs) == 1 {
// sequential job
err = services.SubmitJob(&jobs[0])
if err != nil {
logger.Error("Failed to submit new sequential job.", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil)
return
}
} else {
// array job
arrayJob := models.ArrayJobInfo{
JobID: primaryJobID,
ArrayJobName: arrayJobName,
SubmitUser: submitUser,
QueueName: req.QueueName,
UserGroup: req.UserGroup,
JobDescription: req.JobDescription,
JobName: req.JobName,
ProjectName: req.ProjectName,
LicenseProject: req.LicenseProject,
SLA: req.SLA,
}
err = services.SubmitArrayJobs(arrayJob, jobs)
if err != nil {
logger.Error("Failed to write new array jobs.", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil)
return
}
err = postJobRequest(&job)
if err != nil {
logger.Error("Failed to handle postJobRequest ", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, err.Error(), nil)
return
}
logger.Debug("jobinfo before submit.", zap.Any("job", job))
// Submit job to backend (single call handles both normal and array)
err = services.SubmitJob(&job, arraySize)
if err != nil {
logger.Error("Failed to submit job.", zap.Error(err))
ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil)
return
}
logger.Info("Job submission successfully from ", zap.String("client_ip", ctx.ClientIP()), zap.Any("job", jobs[0]))
logger.Info("Job submission successfully from ", zap.String("client_ip", ctx.ClientIP()), zap.Int64("jobID", primaryJobID))
response := models.JobSubmitResponse{
Message: fmt.Sprintf("Job <%d> is submitted to default queue <%s>.", primaryJobID, req.QueueName),
@ -1548,37 +1519,6 @@ func (j JobController) CreateJob(ctx *gin.Context) {
QueueName: req.QueueName,
}
ReturnSuccess(ctx, common.CREATED, common.SUCCESS_MSG, response)
}
func generateJobIDs(indices []int) ([]models.JobArrayID, error) {
logger.Debug("Enter generateJobIDs() function", zap.Any("indices", indices))
// initialize jobIDPairs
jobIDPairs := make([]models.JobArrayID, 0, len(indices))
// if indices is empty, generate the primary jobid
if len(indices) == 0 {
jobID, err := jobID_Worker.NextID()
if err != nil {
logger.Error("Failed to generate job id.", zap.Error(err))
return nil, fmt.Errorf("failed to generate job id: %w", err)
}
jobIDPairs = append(jobIDPairs, models.JobArrayID{ArrayIdx: 0, Unique_JobID: jobID})
return jobIDPairs, nil
}
// generate the other job IDs based on indices
for _, index := range indices {
jobID, err := jobID_Worker.NextID()
if err != nil {
logger.Error("Failed to generate job id.", zap.Int("index", index), zap.Error(err))
return nil, fmt.Errorf("failed to generate job id: %w", err)
}
jobIDPairs = append(jobIDPairs, models.JobArrayID{ArrayIdx: index, Unique_JobID: jobID})
}
logger.Debug("Generated jobIDPairs", zap.Any("jobIDPairs", jobIDPairs))
return jobIDPairs, nil
}
func preCheckJobRestartRequest(req *models.JobRestartRequest) error {
@ -1589,7 +1529,7 @@ func preCheckJobRestartRequest(req *models.JobRestartRequest) error {
return nil
}
func preCheckJobRequest(req *models.JobSubmitRequest) error {
func preCheckJobRequest(req *models.JobSubmitRequest) ([]int, error) {
logger.Debug("Enter preCheckJobRequest")
if req.QueueName == "" {
req.QueueName = "normal"
@ -1598,33 +1538,32 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error {
queueInfo, ok := store.Queues.Get(req.QueueName)
if !ok {
logger.Error("Invalid queue name specified", zap.String("queue_name", req.QueueName))
return fmt.Errorf("invalid queue name: %s", req.QueueName)
return nil, 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 fmt.Errorf("cannot submit job to queue %s: queue is closed", req.QueueName)
return nil, fmt.Errorf("cannot submit job to queue %s: queue is closed", req.QueueName)
}
}
if req.JobName != "" {
arrayName, parsedIndices, slotLimit, err := createArrayName(req.JobName)
arrayName, idx, slotLimit, err := createArrayName(req.JobName)
if err != nil {
logger.Error("job name format is invalid", zap.Error(err))
return errors.New("job array list format is invalid: " + err.Error())
return nil, errors.New("job array list format is invalid: " + err.Error())
}
if arrayName != "" {
if err = isValidArrayName(arrayName); err != nil {
logger.Error(err.Error())
return errors.New("job array list format is invalid: " + err.Error())
return nil, errors.New("job array list format is invalid: " + err.Error())
}
}
indices = parsedIndices
req.JobName = arrayName
logger.Debug("preCheckJobRequest parsed job name is: ", zap.String("jobName", req.JobName),
zap.Any("indices", indices), zap.Any("slotLimit", slotLimit))
return nil
zap.Any("indices", idx), zap.Any("slotLimit", slotLimit))
return idx, nil
} else {
if req.Command != "" {
if len(req.Command) > 4094 {
@ -1637,24 +1576,24 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error {
if req.JobDescription != "" {
if len(req.JobDescription) > 4094 {
return errors.New("job description length exceeds the limit of 4094 characters")
return nil, 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 err
return nil, err
}
if req.ProjectName != "" {
if len(req.ProjectName) > 511 {
return errors.New("project name length exceeds the limit of 511 characters")
return nil, errors.New("project name length exceeds the limit of 511 characters")
}
}
if req.Resreq != "" {
if len(req.Resreq) > 4094 {
return errors.New("resreq length exceeds the limit of 4094 characters")
return nil, errors.New("resreq length exceeds the limit of 4094 characters")
}
}
@ -1695,16 +1634,16 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error {
// handle environment variables
if req.Env != nil {
if len(req.Env) > 4094 {
return errors.New("environment variable length exceeds the limit of 4094 characters")
return nil, 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 errors.New("environment variable should use key=value format")
return nil, errors.New("environment variable should use key=value format")
}
key := parseEnvs[0]
if strings.TrimSpace(key) == "" {
return errors.New("environment variable key cannot be empty")
return nil, errors.New("environment variable key cannot be empty")
}
}
}
@ -1717,7 +1656,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error {
for _, transferFileStr := range req.TransferFiles {
localFile, operator, remoteFile, err := parseTransferFiles(transferFileStr)
if err != nil {
return err
return nil, err
}
/* based on different operator, check the local file and remote file
@ -1728,7 +1667,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error {
*/
if operator == ">" || operator == "<>" || operator == "><" {
if localFile == "" {
return errors.New("local file is required for operator " + operator)
return nil, errors.New("local file is required for operator " + operator)
}
// Append to realCopyLocalFiles for later processing
realCopyLocalFiles = append(realCopyLocalFiles, localFile)
@ -1736,7 +1675,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error {
if operator == "<" || operator == "<<" {
if remoteFile == "" {
return errors.New("remote file is required for operator " + operator)
return nil, errors.New("remote file is required for operator " + operator)
}
fakeCopyRemoteFile = append(fakeCopyRemoteFile, remoteFile)
}
@ -1751,7 +1690,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error {
// Convert inFileData to JSON and assign to req.InFile
extendDataJSON, err := json.Marshal(extendData)
if err != nil {
return fmt.Errorf("failed to marshal InFile JSON: %v", err)
return nil, fmt.Errorf("failed to marshal InFile JSON: %v", err)
}
req.ExtendData = string(extendDataJSON)
}
@ -1763,7 +1702,7 @@ func preCheckJobRequest(req *models.JobSubmitRequest) 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 fmt.Errorf("invalid NumProcessors specification: %w", err)
return nil, fmt.Errorf("invalid NumProcessors specification: %w", err)
}
req.NumProcessors = minTasks
req.MaxNumProcessors = maxTasks
@ -1772,22 +1711,22 @@ func preCheckJobRequest(req *models.JobSubmitRequest) error {
}
if req.MigThreshold < 0 || req.MigThreshold > 2147483647 {
return errors.New("Incorrect migration threshold value. Specify a number greater than or equal to zero.")
return nil, 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 errors.New("Interactive field must be 0, 1, or 2")
return nil, 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 errors.New("Launch_type must be openmpi or mpich")
return nil, errors.New("Launch_type must be openmpi or mpich")
}
req.LaunchType = launchType
}
return nil
return nil, nil
}
// It returns minTasks, maxTasks, and an error if parsing fails or constraints are violated.