package jobs import ( "encoding/json" "errors" "fmt" "io" "os" "os/exec" "regexp" "runtime" "strings" "time" "xcecli/common/format" "xcecli/common/request" "xcecli/common/utils" "xcecli/config" logger "xcecli/config" "xcecli/models" "github.com/spf13/cobra" "github.com/spf13/pflag" "go.uber.org/zap" ) // JobSubmitter structure for managing job submission from the CLI parameters and operations type JobSubmitter struct { configManager *config.ConfigManager queue string resReq string beginTime string termTime string jobName string jobDescription string cpuLimit string command string runLimit string coreLimit string fileLimit string memLimit string swapLimit string dataLimit string processLimit string stackLimit string threadLimit string useShellLimit bool useSystemEnv bool subEnvVars []string cwd string uploadFiles []string processorRange string infile string outfile string errfile string outdir string configFile string interactive bool useTTY bool useXJob bool projectName string askedHosts string launchType string preExecCmd string postExecCmd string } // NewJobSubmitter creates a new JobSubmitter instance func NewJobSubmitter(configManager *config.ConfigManager) *JobSubmitter { return &JobSubmitter{ configManager: configManager, } } // Parse parses command line arguments into JobSubmitter struct func (js *JobSubmitter) ParseCmdArgs(cmd *cobra.Command, args []string) error { // check if the -R flag is used more than once if cmd.Flags().Changed("resreq") { count := 0 cmd.Flags().Visit(func(f *pflag.Flag) { if f.Name == "resreq" { count++ } }) if count > 1 { return fmt.Errorf("error: the -R/--resreq flag can only be used once") } } if js.interactive { if js.useTTY && js.useXJob { return fmt.Errorf("error: cannot use both --tty and --xjob") } } // check if there is standard input redirection fileInfo, err := os.Stdin.Stat() if err == nil && (fileInfo.Mode()&os.ModeCharDevice) == 0 { if tmpFileName, err := js.handleStdin(); err != nil { errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("error reading from stdin: [%s]", errMsg) } else if tmpFileName != "" { defer os.Remove(tmpFileName) } } // if the config file is specified, parse the config file first if js.configFile != "" { return js.parseConfigFile() } // command line parameters will override the settings in the config file if len(args) > 0 { js.command = strings.Join(args, " ") } return nil } func (js *JobSubmitter) handleStdin() (string, error) { content, err := io.ReadAll(os.Stdin) if err != nil { return "", err } if len(content) == 0 { return "", nil } tmpFile, err := os.CreateTemp("", "xcecli_stdin_*.xce") if err != nil { errMsg := request.ExtractErrorMessage(err) return "", fmt.Errorf("failed to create temporary file: [%s]", errMsg) } defer tmpFile.Close() if _, err := tmpFile.Write(content); err != nil { errMsg := request.ExtractErrorMessage(err) return "", fmt.Errorf("failed to write to temporary file: [%s]", errMsg) } js.configFile = tmpFile.Name() return js.configFile, nil } // parseConfigFile parse the parameters in the config file func (js *JobSubmitter) parseConfigFile() error { // check if the file exists fileInfo, err := os.Stat(js.configFile) if err != nil { errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("config file %s does not exist or cannot be accessed: [%s]", js.configFile, errMsg) } if fileInfo.IsDir() { return fmt.Errorf("%s is a directory, not a valid config file", js.configFile) } // if the file is a temporary file, delete it when the function ends if strings.Contains(js.configFile, "xcecli_stdin_") { defer os.Remove(js.configFile) } // read the config file content content, err := os.ReadFile(js.configFile) if err != nil { errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("failed to read the config file: %s", errMsg) } lines := strings.Split(string(content), "\n") fmt.Printf("config file %s contains %d lines of commands\n", js.configFile, len(lines)) // create the main command rootCmd := NewSubmitJobsCmd(js.configManager) // collect all parameters and commands var allParams []string var allCmdParts []string // process each line for _, line := range lines { line = strings.TrimSpace(line) if line == "" { continue // skip empty lines } if strings.HasPrefix(line, "#xce job submit") { // process the parameters in the comment line paramLine := strings.TrimPrefix(line, "#xce job submit") paramLine = strings.TrimSpace(paramLine) if paramLine != "" { parts := strings.Fields(paramLine) // separate the parameters and command parts for j := 0; j < len(parts); j++ { if strings.HasPrefix(parts[j], "-") { allParams = append(allParams, parts[j]) // if the parameter has a value and is not a - if j+1 < len(parts) && !strings.HasPrefix(parts[j+1], "-") { allParams = append(allParams, parts[j+1]) j++ } } else { // non-parameter part as command cmdPart := strings.Join(parts[j:], " ") if cmdPart != "" { allCmdParts = append(allCmdParts, cmdPart) } break } } } } else { // non-comment line as command part allCmdParts = append(allCmdParts, line) } } // merge all command parts fullCommand := strings.Join(allCmdParts, " ") // merge all parameters fullArgs := append(allParams, strings.Fields(fullCommand)...) fmt.Println("merged full command:") fmt.Printf("xce jobs submit %s\n", strings.Join(fullArgs, " ")) // execute the merged command rootCmd.SetArgs(fullArgs) if err := rootCmd.Execute(); err != nil { errMsg := request.ExtractErrorMessage(err) return fmt.Errorf("failed to execute the command: [%s]", errMsg) } os.Exit(0) return nil } // BuildJobRequest func (js *JobSubmitter) BuildJobRequest() (*models.JobSubmitRequest, error) { // create the job request with only required fields jobReq := &models.JobSubmitRequest{ Command: js.command, } if js.interactive { if js.useTTY { jobReq.Interactive = 1 // TTY mode } else if js.useXJob { jobReq.Interactive = 2 // XJob mode } else { jobReq.Interactive = 3 // only interactive, no special mode } } else { jobReq.Interactive = 0 // default value } // Helper function to format limit values with space formatLimit := func(value string) string { if value == "" { return "" } re := regexp.MustCompile(`^(\d+)(MB|GB|TB|PB|[MGTP])$`) matches := re.FindStringSubmatch(strings.ToUpper(value)) if matches != nil { return matches[1] + " " + matches[2] } return value } // Set resource limits jobReq.SwapLimit = formatLimit(js.swapLimit) jobReq.DataLimit = formatLimit(js.dataLimit) jobReq.MemLimit = formatLimit(js.memLimit) jobReq.FileLimit = formatLimit(js.fileLimit) jobReq.CoreLimit = formatLimit(js.coreLimit) jobReq.ProcessLimit = formatLimit(js.processLimit) jobReq.StackLimit = formatLimit(js.stackLimit) jobReq.ThreadLimit = formatLimit(js.threadLimit) jobReq.CPULimit = formatLimit(js.cpuLimit) jobReq.RunLimit = formatLimit(js.runLimit) // Set other fields if explicitly set jobReq.QueueName = js.queue jobReq.Resreq = js.resReq jobReq.JobName = js.jobName jobReq.JobDescription = js.jobDescription // Parse processor range if specified if js.processorRange != "" { parts := strings.Split(js.processorRange, ",") if len(parts) == 2 { // Try to parse min and max processors minProc, err := utils.ParseInt(parts[0]) if err != nil { return nil, errors.New("invalid min processor value") } maxProc, err := utils.ParseInt(parts[1]) if err != nil { return nil, errors.New("invalid max processor value") } jobReq.NumProcessors = minProc jobReq.MaxNumProcessors = maxProc } else { // Single value case numProc, err := utils.ParseInt(js.processorRange) if err != nil { return nil, errors.New("invalid processor value") } jobReq.NumProcessors = numProc } } // parse the begin time if js.beginTime != "" { // Parse the time in local timezone parsedTime, err := utils.ParseTime(js.beginTime) if err != nil { errMsg := request.ExtractErrorMessage(err) return nil, fmt.Errorf("invalid begin time value and %s", errMsg) } jobReq.BeginTime = parsedTime } // If the date is in the past, add appropriate time // parse the end time if js.termTime != "" { parsedTime, err := utils.ParseTime(js.termTime) if err != nil { errMsg := request.ExtractErrorMessage(err) return nil, fmt.Errorf("invalid end time value and %s", errMsg) } jobReq.TermTime = parsedTime } // if shell limit is enabled, get the system limits if js.useShellLimit { limits := utils.NewSystemLimits() // set the limits in the job request if val, err := limits.GetCPULimit(); err == nil && val > 0 { jobReq.CPULimit = fmt.Sprintf("%d", val) } if val, err := limits.GetFileLimit(); err == nil && val > 0 { jobReq.FileLimit = fmt.Sprintf("%d", val) } if val, err := limits.GetDataLimit(); err == nil && val > 0 { jobReq.DataLimit = fmt.Sprintf("%d", val) } if val, err := limits.GetStackLimit(); err == nil && val > 0 { jobReq.StackLimit = fmt.Sprintf("%d", val) } if val, err := limits.GetCoreLimit(); err == nil && val > 0 { jobReq.CoreLimit = fmt.Sprintf("%d", val) } if val, err := limits.GetProcessLimit(); err == nil && val > 0 { jobReq.ProcessLimit = fmt.Sprintf("%d", val) } if val, err := limits.GetThreadLimit(); err == nil && val > 0 { jobReq.ThreadLimit = fmt.Sprintf("%d", val) } if val, err := limits.GetMemLimit(); err == nil && val > 0 { jobReq.MemLimit = fmt.Sprintf("%d", val) } if val, err := limits.GetSwapLimit(); err == nil && val > 0 { jobReq.SwapLimit = fmt.Sprintf("%d", val) } } // if system environment variables are enabled, get and set them if js.useSystemEnv { systemEnv := utils.GetSystemEnv() filteredEnv := []string{} filteredEnv = append(filteredEnv, systemEnv...) jobReq.Env = filteredEnv } // Add submitted environment variables if len(js.subEnvVars) > 0 { jobReq.Env = append(jobReq.Env, js.getEnvironmentVariables()...) } if js.infile != "" { jobReq.InFile = js.infile } if js.outfile != "" { jobReq.OutFile = js.outfile } if js.errfile != "" { jobReq.ErrFile = js.errfile } if js.outdir != "" { jobReq.OutDir = js.outdir } if js.projectName != "" { jobReq.ProjectName = js.projectName } if js.cwd != "" { jobReq.CWD = js.cwd } else { cwd, _ := os.Getwd() jobReq.CWD = cwd } if js.askedHosts != "" { jobReq.AskedHosts = js.askedHosts } if js.launchType != "" { jobReq.LaunchType = js.launchType } if js.preExecCmd != "" { jobReq.PreExecCmd = js.preExecCmd } if js.postExecCmd != "" { jobReq.PostExecCmd = js.postExecCmd } return jobReq, nil } // Run executes the job submission operation func (js *JobSubmitter) RunJobsSubmit() error { // Get server info and validate authentication serverInfo, _ := js.configManager.GetServerInfo() if js.command == "" { return fmt.Errorf("please provide the command to execute") } if js.infile != "" { js.uploadFiles = append(js.uploadFiles, fmt.Sprintf("%s > %s", js.infile, js.infile)) } if js.outfile != "" { js.uploadFiles = append(js.uploadFiles, fmt.Sprintf("%s < %s", js.outfile, js.outfile)) } if js.errfile != "" { js.uploadFiles = append(js.uploadFiles, fmt.Sprintf("%s < %s", js.errfile, js.errfile)) } fileExists := func(path string) error { info, err := os.Stat(path) if err != nil { return fmt.Errorf("error: the file %s does not exist, please check the file name or path", path) } if info.IsDir() { return fmt.Errorf("error: the file %s is a directory, please check the file name or path", path) } return nil } if js.infile != "" { if err := fileExists(js.infile); err != nil { return err } } if js.outdir != "" { info, err := os.Stat(js.outdir) if os.IsNotExist(err) { return fmt.Errorf("error: the output directory %s does not exist", js.outdir) } if !info.IsDir() { return fmt.Errorf("error: the output directory %s is not a directory", js.outdir) } } // create the API client and submit the job apiClient := request.NewAPIClient(serverInfo.URL) var transferFile []string if len(js.uploadFiles) > 0 { fileTrans := newFileTransfer(js.uploadFiles) err := fileTrans.uploadFiles(func(s []string) (*models.UploadResponse, error) { return apiClient.PostFile("Bearer "+serverInfo.Token, s) }) if err != nil { return err } transferFile = fileTrans.buildTransferFileList() } // build the job request jobReq, err := js.BuildJobRequest() if err != nil { return err } jobReq.TransferFiles = transferFile jobResp, err := apiClient.SubmitJob("Bearer "+serverInfo.Token, jobReq) logger.Debug("jobResp", zap.Any("jobResp", jobResp), zap.Error(err)) logger.Info("jobResp", zap.Any("jobResp", jobResp), zap.Error(err)) if err != nil { return err } msg := strings.Replace(jobResp.Data.Message, fmt.Sprintf("<%d>", jobResp.Data.JobID), fmt.Sprintf("<%s>", format.FormatJobID(jobResp.Data.JobID)), 1) fmt.Println(msg) logger.Debug("job submitted successfully", zap.String("message", jobResp.Data.Message)) logger.Info("job submitted successfully", zap.String("message", jobResp.Data.Message)) if js.interactive { fmt.Println("<>") url, err := js.waitForInteractiveURL(apiClient, serverInfo.Token, fmt.Sprintf("%d", jobResp.Data.JobID)) if err != nil { return err } fmt.Printf("interactive session URL: %s\n", url) // automatically open the browser if err := openBrowser(url); err != nil { errMsg := request.ExtractErrorMessage(err) fmt.Printf("warning: failed to open browser: [%s]\n", errMsg) fmt.Println("Please manually open the URL in your browser.") } } return nil } func (js *JobSubmitter) checkJobStatus(apiClient *request.APIClient, token string, jobID string) (string, error) { var resp models.JobsResponse err := apiClient.GetRequestFromAPIServer( "Bearer "+token, "", fmt.Sprintf("/xce/v1/jobs/%s", jobID), &resp, map[string]string{ /*"fields": "interactive_job_url"*/ }, ) if err != nil { errMsg := request.ExtractErrorMessage(err) return "", fmt.Errorf("failed to get job details: [%s]", errMsg) } for _, row := range resp.Data.Rows { switch row.JobStatus { case "PEND": return row.JobStatus, nil case "RUN": var execHosts models.ExecHosts if err := json.Unmarshal([]byte(row.ExecHosts.String()), &execHosts); err == nil && len(execHosts.HostSlots) > 0 { fmt.Println("<>") } return row.JobStatus, nil default: return "", fmt.Errorf("job %s is in state %s, please check the job status", jobID, row.JobStatus) } } return "", fmt.Errorf("timeout waiting for job status") } func (js *JobSubmitter) checkExtendData(apiClient *request.APIClient, token string, jobID string) (string, error) { var resp models.JobsResponse err := apiClient.GetRequestFromAPIServer( "Bearer "+token, "", fmt.Sprintf("/xce/v1/jobs/%s", jobID), &resp, map[string]string{"fields": "interactive_job_url"}, ) if err != nil { errMsg := request.ExtractErrorMessage(err) return "", fmt.Errorf("failed to get job details: [%s]", errMsg) } for _, row := range resp.Data.Rows { //fmt.Println(row) if row.InteractiveJobURL != "" { return row.InteractiveJobURL, nil } } return "", fmt.Errorf("no interactive job URL found") } func (js *JobSubmitter) waitForInteractiveURL(apiClient *request.APIClient, token string, jobID string) (string, error) { timeout := time.After(24 * time.Hour) timer := time.NewTimer(0) defer timer.Stop() for { select { case <-timeout: return "", fmt.Errorf("timeout waiting for interactive URL") case <-timer.C: status, err := js.checkJobStatus(apiClient, token, jobID) if err != nil { return "", err } if status == "PEND" { timer.Reset(5 * time.Second) continue } if status == "RUN" { //time.Sleep(45 * time.Second) return js.checkExtendData(apiClient, token, jobID) } return "", fmt.Errorf("job %s is in state %s, please check the job status", jobID, status) } } } func openBrowser(url string) error { var err error switch runtime.GOOS { case "linux": err = exec.Command("xdg-open", url).Start() case "windows": err = exec.Command("cmd", "/c", "start", url).Start() case "darwin": err = exec.Command("open", url).Start() default: err = fmt.Errorf("unsupported platform") } return err } // NewSubCmd creates a command for submitting jobs func NewSubmitJobsCmd(configManager *config.ConfigManager) *cobra.Command { jobSubmitter := NewJobSubmitter(configManager) cmd := &cobra.Command{ Use: "submit", Short: "Submit job to API Server", Long: `Submit job to API Server. Example: xce jobs submit -q queue_name sleep 10 # submit job to queue queue_name with command "sleep 10" xce jobs submit -R "select(!mg)" sleep 10 # submit job with resource requirement "select(!mg)" xce jobs submit -c 10 sleep 10 # submit job with cpu limit 10 xce jobs submit -C 10 sleep 10 # submit job with core limit 10 xce jobs submit -b "2025-03-20 10:00" sleep 10 # submit job with begin time "2025-03-20 10:00" xce jobs submit -t "2025-03-20 10:00" sleep 10 # submit job with termination time "2025-03-20 10:00" xce jobs submit -J "test" sleep 10 # submit job with job name "test" xce jobs submit --Jd "test" sleep 10 # submit job with job description "test" xce jobs submit -F 1024 sleep 10 # submit job with file limit 1024 xce jobs submit -W 10 sleep 10 # sets the runtime limit of the job xce jobs submit -M 1024 sleep 10 # sets the memory limit of the job xce jobs submit -v 1024 sleep 10 # sets the swap limit of the job xce jobs submit -D 1024 sleep 10 # sets the data limit of the job xce jobs submit -p 10 sleep 10 # sets the process limit of the job xce jobs submit -S 1024 sleep 10 # sets the stack limit of the job xce jobs submit -T 1024 sleep 10 # sets the thread limit of the job xce jobs submit --ul sleep 10 # use current shell limits xce jobs submit --env sleep 10 # use system environment variables xce jobs submit --cwd "/tmp/" sleep 10 # use current working directory for job execution xce jobs submit -f "t1.txt > t2.txt" sleep 10 # submit job with file "t1.txt" and redirect the output to "t2.txt" xce jobs submit -q "normal" sleep 10 # submit job to queue "normal" and "priority" xce jobs submit -n 4,6 sleep 10 # submit job with processor range 4-6 xce jobs submit -P project_name sleep 10 # submit job with project name xce jobs submit --configfile job.xce # submit job using parameters from a configuration file xce jobs submit -m "host1 host2" sleep 10 # submit job into host1 host2 xce jobs submit -I --tty sleep 10 # submit interactive job with tty xce jobs submit -I --xjob sleep 10 # submit interactive job with xjob`, SilenceUsage: true, Run: func(cmd *cobra.Command, args []string) { if err := jobSubmitter.ParseCmdArgs(cmd, args); err != nil { fmt.Println(strings.ToLower(err.Error())) return } if err := jobSubmitter.RunJobsSubmit(); err != nil { fmt.Println(strings.ToLower(err.Error())) return } }, } // add the flags flags := cmd.Flags() // modify the resreq flag resReqFlag := &pflag.Flag{ Name: "resreq", Shorthand: "R", Usage: "specify the resource requirement (can only be used once)", Value: (*stringValue)(&jobSubmitter.resReq), NoOptDefVal: "", } flags.AddFlag(resReqFlag) //modify the flags flags.StringVarP(&jobSubmitter.beginTime, "begin_time", "b", "", "specify the begin time") flags.StringVarP(&jobSubmitter.coreLimit, "core_limit", "C", "", "specify the core number") flags.StringVarP(&jobSubmitter.jobName, "job_name", "J", "", "specify the job name") flags.StringVarP(&jobSubmitter.jobDescription, "Jd", "", "", "specify the job description") flags.StringVarP(&jobSubmitter.termTime, "term_time", "t", "", "specify the termination time") flags.StringVarP(&jobSubmitter.cpuLimit, "cpu_limit", "c", "", "specify the cpu limit") flags.StringVarP(&jobSubmitter.runLimit, "run_limit", "W", "", "specify the run time") flags.StringVarP(&jobSubmitter.fileLimit, "file_limit", "F", "", "specify the file limit") flags.StringVarP(&jobSubmitter.memLimit, "mem_limit", "M", "", "specify the memory limit") flags.StringVarP(&jobSubmitter.swapLimit, "swap_limit", "v", "", "specify the swap limit") flags.StringVarP(&jobSubmitter.dataLimit, "data_limit", "D", "", "specify the data limit") flags.StringVarP(&jobSubmitter.processLimit, "process_limit", "p", "", "specify the process limit") flags.StringVarP(&jobSubmitter.stackLimit, "stack_limit", "S", "", "specify the stack limit") flags.StringVarP(&jobSubmitter.threadLimit, "thread_limit", "T", "", "specify the thread limit") flags.BoolVarP(&jobSubmitter.useShellLimit, "ul", "", false, "use current shell limits") flags.BoolVarP(&jobSubmitter.useSystemEnv, "env", "", false, "use system environment variables") flags.StringVarP(&jobSubmitter.cwd, "cwd", "", "", "current working directory") flags.StringArrayVarP(&jobSubmitter.uploadFiles, "file", "f", []string{}, "specify the files to upload") flags.StringVarP(&jobSubmitter.queue, "queue_name", "q", "", "specify the job queue") flags.StringVarP(&jobSubmitter.processorRange, "processor_range", "n", "", "specify processor range in format min,max (e.g. 4,6)") flags.StringVarP(&jobSubmitter.infile, "infile", "i", "", "specify the input file") flags.StringVarP(&jobSubmitter.outfile, "outfile", "o", "", "specify the output file") flags.StringVarP(&jobSubmitter.errfile, "errfile", "e", "", "specify the error file") flags.StringVarP(&jobSubmitter.outdir, "outdir", "d", "", "specify the output directory") flags.StringVarP(&jobSubmitter.projectName, "project_name", "P", "", "specify the project name") flags.StringVar(&jobSubmitter.configFile, "configfile", "", "specify the configuration file for job submission") flags.BoolVarP(&jobSubmitter.interactive, "interactive", "I", false, "submit interactive job") flags.BoolVar(&jobSubmitter.useTTY, "tty", false, "use TTY for interactive job") flags.BoolVar(&jobSubmitter.useXJob, "xjob", false, "use XJob for interactive job") flags.StringVarP(&jobSubmitter.askedHosts, "m", "m", "", "specify the running hosts, with space for multiple values") flags.StringVarP(&jobSubmitter.launchType, "launch_type", "", "", "specify the launch type") flags.StringVarP(&jobSubmitter.preExecCmd, "pre_exec", "", "", "specify the pre-execution command") flags.StringVarP(&jobSubmitter.postExecCmd, "post_exec", "", "", "specify the post-execution command") return cmd // } // modify the stringValue type stringValue string func (s *stringValue) String() string { return string(*s) } func (s *stringValue) Set(val string) error { if *s != "" { return fmt.Errorf("the -R/--resreq flag can only be used once") } *s = stringValue(val) return nil } func (s *stringValue) Type() string { return "string" } func (js *JobSubmitter) getEnvironmentVariables() []string { var envVars []string for _, env := range js.subEnvVars { // Skip environment variables with multiple equal signs if strings.Count(env, "=") > 1 { continue } envVars = append(envVars, env) } return envVars }