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

fix(apiserver): restart carries only the options a job gets (bug-1968)

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-10-10 14:23:16 +00:00
parent 801d26db06
commit 6a2c69f8bd
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
2 changed files with 158 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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 != "" {

125
src/frontend/apiserver/controllers/jobs_handler_test.go
View file

@ -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,