From 6a2c69f8bd2f7d3b53ad63579516ceb2b40fcc22 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 10 Oct 2026 14:23:16 +0000 Subject: [PATCH] fix(apiserver): restart carries only the options a job gets (bug-1968) RestartJobByID rebuilt the new job from the old job's row: - it accepted asked_hosts and copied asked hosts, I/O paths and out_dir, which no job gets (refused at submit since bug-1936); restart now refuses asked_hosts with 400 and no longer carries those over; - it read the JSONB columns chkpnt_info/extend_data as strings, but rows keep models.Jsonb, so env/limits were lost and a checkpointed job never restarted from its checkpoint; new rowJSON() reads both; - it read "res_req" (the column is "resreq"), losing the resource requirement, and never copied exec_home. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/frontend/apiserver/controllers/jobs.go | 78 +++++------ .../controllers/jobs_handler_test.go | 125 ++++++++++++++++++ 2 files changed, 158 insertions(+), 45 deletions(-) diff --git a/src/frontend/apiserver/controllers/jobs.go b/src/frontend/apiserver/controllers/jobs.go index ce65599e..7e13c56a 100644 --- a/src/frontend/apiserver/controllers/jobs.go +++ b/src/frontend/apiserver/controllers/jobs.go @@ -857,7 +857,7 @@ func (j JobController) RestartJobByID(ctx *gin.Context) { err = preCheckJobRestartRequest(&req) if err != nil { logger.Error("Failed to restart job", zap.Error(err)) - ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) + ReturnError(ctx, common.INVALID_PARAMS_CODE, err.Error(), nil) return } @@ -918,18 +918,11 @@ func (j JobController) RestartJobByID(ctx *gin.Context) { JobStatus: models.JOBSTAT["PEND"], } - err = processJobRestartRequest(&newJob, &req) - if err != nil { - logger.Error("Failed to restart job", zap.Error(err)) - ReturnError(ctx, common.ERROR_CODE, err.Error(), nil) - return - } - if v, ok := oldJob["job_name"].(string); ok { newJob.JobName = v } - if chkpntStr, ok := oldJob["chkpnt_info"].(string); ok && chkpntStr != "" { + if chkpntStr := rowJSON(oldJob["chkpnt_info"]); chkpntStr != "" { var chkpntInfo map[string]interface{} if err := json.Unmarshal([]byte(chkpntStr), &chkpntInfo); err == nil { if restartCmd, ok := chkpntInfo["restart_command"].(string); ok { @@ -938,7 +931,7 @@ func (j JobController) RestartJobByID(ctx *gin.Context) { } if ckpt_pid, ok := chkpntInfo["ckpt_pid"].(float64); ok { extendData := map[string]interface{}{} - if v, ok := oldJob["extend_data"].(string); ok && v != "" { + if v := rowJSON(oldJob["extend_data"]); v != "" { _ = json.Unmarshal([]byte(v), &extendData) } if v, ok := extendData["sub_env_vars"]; ok && v != nil { @@ -953,10 +946,6 @@ func (j JobController) RestartJobByID(ctx *gin.Context) { } extendDataBytes, _ := json.Marshal(extendData) newJob.ExtendData = models.Jsonb{String: string(extendDataBytes)} - } else { - if v, ok := oldJob["extend_data"].(string); ok && v != "" { - newJob.ExtendData = models.Jsonb{String: v} - } } } } else { @@ -984,7 +973,7 @@ func (j JobController) RestartJobByID(ctx *gin.Context) { if v, ok := oldJob["job_description"].(string); ok { newJob.JobDescription = v } - if v, ok := oldJob["res_req"].(string); ok { + if v, ok := oldJob["resreq"].(string); ok { newJob.ResReq = v } if v, ok := oldJob["rsv_id"].(string); ok { @@ -999,17 +988,6 @@ func (j JobController) RestartJobByID(ctx *gin.Context) { if v, ok := oldJob["num_processors"].(int); ok { newJob.NumProcessors = v } - if newJob.AskedHosts == "" { - if v, ok := oldJob["asked_hosts"].(string); ok { - newJob.AskedHosts = v - } - } - if v, ok := oldJob["begin_time"].(time.Time); ok { - newJob.BeginTime = models.CustomTime{Time: v} - } - if v, ok := oldJob["term_time"].(time.Time); ok { - newJob.TermTime = models.CustomTime{Time: v} - } if v, ok := oldJob["job_priority"].(int); ok { newJob.JobPriority = v } @@ -1046,17 +1024,17 @@ func (j JobController) RestartJobByID(ctx *gin.Context) { if v, ok := oldJob["chkpnt_dir"].(string); ok { newJob.ChkpntDir = v } - if v, ok := oldJob["out_dir"].(string); ok { - newJob.OutDir = v + // bug-1968: asked hosts, begin/term time, in/out/err files and out_dir + // are not carried over: no job gets them (refused at submit since + // bug-1936), so the restarted job would only show them. Extended data + // (env, limits) and exec home are, checkpointed or not. + if newJob.ExtendData.String == "" { + if v := rowJSON(oldJob["extend_data"]); v != "" { + newJob.ExtendData = models.Jsonb{String: v} + } } - if v, ok := oldJob["in_file"].(string); ok { - newJob.InFile = v - } - if v, ok := oldJob["err_file"].(string); ok { - newJob.ErrFile = v - } - if v, ok := oldJob["out_file"].(string); ok { - newJob.OutFile = v + if v, ok := oldJob["exec_home"].(string); ok { + newJob.ExecHome = v } if v, ok := oldJob["login_shell"].(string); ok { newJob.LoginShell = v @@ -1347,14 +1325,31 @@ func preCheckSubmitOptions(req *models.JobSubmitRequest) error { return nil } +// preCheckJobRestartRequest refuses a restart option the restarted job would +// not get (bug-1968, as preCheckSubmitOptions does for submit, bug-1936): +// asked hosts have no field in the controller's submit data. func preCheckJobRestartRequest(req *models.JobRestartRequest) error { logger.Debug("Enter preCheckJobRestartRequest") - if err := validateAskedHosts(req.AskedHosts); err != nil { - return err + if strings.TrimSpace(req.AskedHosts) != "" { + return fmt.Errorf("job restart option asked_hosts not supported: the job would not get it") } return nil } +// rowJSON returns a JSON column of a stored job row as a string: rows come +// from store.StructToOrderMap, which keeps each field's Go type, so a JSONB +// column (chkpnt_info, extend_data) is a models.Jsonb, not a string +// (bug-1968). +func rowJSON(v interface{}) string { + switch j := v.(type) { + case models.Jsonb: + return j.String + case string: + return j + } + return "" +} + // validateResreqSyntax is the resource-request syntax check used by // preCheckJobRequest; a variable so tests can substitute it (bug-1744). var validateResreqSyntax = backend.ValidateResreqSyntax @@ -1909,13 +1904,6 @@ func parseTransferFiles(transferFileStr string) (localFile, operator, remoteFile return localFile, operator, remoteFile, nil } -func processJobRestartRequest(newJob *models.JobInfo, req *models.JobRestartRequest) error { - logger.Debug("Enter processJobRestartRequest function", zap.Any("req", req)) - if req.AskedHosts != "" { - newJob.AskedHosts = req.AskedHosts - } - return nil -} func processJobRequest(ctx *gin.Context, req *models.JobSubmitRequest, defaultValues map[string]string) error { if req.InFile != "" { diff --git a/src/frontend/apiserver/controllers/jobs_handler_test.go b/src/frontend/apiserver/controllers/jobs_handler_test.go index 23144ac9..3dc1811a 100644 --- a/src/frontend/apiserver/controllers/jobs_handler_test.go +++ b/src/frontend/apiserver/controllers/jobs_handler_test.go @@ -1522,6 +1522,131 @@ func TestJobHandler_RestartNamedJob(t *testing.T) { } } +// bug-1968: restart, like submit (bug-1936), refuses an option the restarted +// job would never get: asked hosts (-m) have no field in the controller's +// submit data. Nothing is submitted. +func TestJobHandler_RestartRejectsAskedHosts(t *testing.T) { + f := newHandlerFixture(t) + ord := handlerTestOrdinaryBase + 50<<15 + f.seedRestartJob(ord, "sleep 1") + + before := len(f.submitted) + w := f.do(http.MethodPost, "/xce/v1/jobs/"+strconv.FormatUint(ord, 10)+"/restart", + map[string]interface{}{"asked_hosts": "hostA"}) + if w.Code != http.StatusBadRequest { + t.Errorf("status = %d, want 400; body: %s", w.Code, w.Body.String()) + } + if len(f.submitted) != before { + t.Errorf("%d submits, want none", len(f.submitted)-before) + } +} + +// bug-1968: the restarted job is rebuilt from the old job's row. Options no +// job can get (asked hosts, begin/term time, I/O paths, out_dir; refused at +// submit since bug-1936) must not be carried over, while the options the job +// does get -- extended data (env, limits), cwd and exec home -- must be, +// whether or not the old job was checkpointed. +func TestJobHandler_RestartCarriesOnlyHonouredOptions(t *testing.T) { + f := newHandlerFixture(t) + ord := handlerTestOrdinaryBase + 51<<15 + exdata := `{"sub_env_vars":{"A":"1"},"limits":{"run_limit":60}}` + f.seedJob(models.JobInfo{ + Unique_JobID: int64(ord), + JobID: int64(backend.ExtractArrayBaseID(ord)), + JobName: "r", + Command: "sleep 1", + JobStatus: models.JOBSTAT["DONE"], + CWD: "/data/run", + ExecHome: "/home/u", + ResReq: "rusage[mem=100]", + ExtendData: models.Jsonb{String: exdata}, + AskedHosts: "hostA", + BeginTime: models.CustomTime{Time: time.Unix(1700000000, 0)}, + TermTime: models.CustomTime{Time: time.Unix(1800000000, 0)}, + InFile: "/in", + OutFile: "/out", + ErrFile: "/err", + OutDir: "/outdir", + }) + + before := len(f.submitted) + w := f.restart(strconv.FormatUint(ord, 10)) + if w.Code != http.StatusOK && w.Code != http.StatusCreated { + t.Fatalf("status = %d; body: %s", w.Code, w.Body.String()) + } + if len(f.submitted) != before+1 { + t.Fatalf("%d submits, want 1", len(f.submitted)-before) + } + job := f.submitted[before].Job + + dropped := map[string]bool{ + "asked_hosts": job.AskedHosts != "", + "begin_time": !job.BeginTime.Time.IsZero(), + "term_time": !job.TermTime.Time.IsZero(), + "in_file": job.InFile != "", + "out_file": job.OutFile != "", + "err_file": job.ErrFile != "", + "out_dir": job.OutDir != "", + } + for field, kept := range dropped { + if kept { + t.Errorf("restarted job kept %s, which no job gets", field) + } + } + if job.ExtendData.String != exdata { + t.Errorf("ExtendData = %q, want %q", job.ExtendData.String, exdata) + } + if job.CWD != "/data/run" { + t.Errorf("CWD = %q, want /data/run", job.CWD) + } + if job.ExecHome != "/home/u" { + t.Errorf("ExecHome = %q, want /home/u", job.ExecHome) + } + if job.ResReq != "rusage[mem=100]" { + t.Errorf("ResReq = %q, want rusage[mem=100]", job.ResReq) + } +} + +// bug-1968: a checkpointed job restarts from its checkpoint: the restart +// command replaces the command and the checkpointed pid reaches the job as +// XCE_JOB_CHECKPOINT_PID next to the old job's environment. The stored +// chkpnt_info/extend_data are models.Jsonb, which the restart never read. +func TestJobHandler_RestartFromCheckpoint(t *testing.T) { + f := newHandlerFixture(t) + ord := handlerTestOrdinaryBase + 52<<15 + f.seedJob(models.JobInfo{ + Unique_JobID: int64(ord), + JobID: int64(backend.ExtractArrayBaseID(ord)), + JobName: "r", + Command: "sleep 1", + JobStatus: models.JOBSTAT["EXIT"], + ChkpntInfo: models.Jsonb{String: `{"restart_command":"criu restore","ckpt_pid":4321}`}, + ExtendData: models.Jsonb{String: `{"sub_env_vars":{"A":"1"}}`}, + }) + + before := len(f.submitted) + w := f.restart(strconv.FormatUint(ord, 10)) + if w.Code != http.StatusOK && w.Code != http.StatusCreated { + t.Fatalf("status = %d; body: %s", w.Code, w.Body.String()) + } + if len(f.submitted) != before+1 { + t.Fatalf("%d submits, want 1", len(f.submitted)-before) + } + job := f.submitted[before].Job + if job.Command != "criu restore" { + t.Errorf("Command = %q, want the checkpoint's restart command", job.Command) + } + var ex struct { + Env map[string]string `json:"sub_env_vars"` + } + if err := json.Unmarshal([]byte(job.ExtendData.String), &ex); err != nil { + t.Fatalf("ExtendData %q: %v", job.ExtendData.String, err) + } + if ex.Env["A"] != "1" || ex.Env["XCE_JOB_CHECKPOINT_PID"] != "4321" { + t.Errorf("sub_env_vars = %v, want A=1 and XCE_JOB_CHECKPOINT_PID=4321", ex.Env) + } +} + // bug-1936: the job handed to the backend carries the cwd and the extended // data under the key names and units the worker's parse_exdata()/ // parse_limits() read (run_limit in seconds, memory limits in MB,