fix(apiserver): send exdata, exec cwd and exec home with submitted jobs (bug-1936)
Cause: modelJobToSubmit() filled only cmdline, execuser, launch_type and pre/post-exec of job_startinfo_t, so the extended data the submit handler builds (env, *_limit incl. -W run limit, file transfers, interactive mode), the cwd and the exec home never reached the controller or the worker. Fix: map the start info in a pure submitStartInfo() (exdata from ExtendData, execcwd from ExecCWD or CWD, exechome from ExecHome) that modelJobToSubmit copies into the C struct; decode exdata back in cJobToModel/cJobArrayToModels so the store keeps ExtendData and CWD. Reject extend_data the worker's parse_exdata() cannot take (not a JSON object, over EXDATA_LENGTH-1 bytes) and drop an empty transfer list instead of sending a null. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
7a9d75d661
commit
d01ced9bd0
5 changed files with 236 additions and 14 deletions
|
|
@ -32,6 +32,15 @@ func goString(s *C.char) string {
|
|||
return C.GoString(s)
|
||||
}
|
||||
|
||||
// cStringOrNil returns a C copy of s, or nil for an empty string so an unset
|
||||
// field stays NULL in the C struct. The caller frees the result.
|
||||
func cStringOrNil(s string) *C.char {
|
||||
if s == "" {
|
||||
return nil
|
||||
}
|
||||
return C.CString(s)
|
||||
}
|
||||
|
||||
// cJobArrayToModels decodes a compact Data.jobArray push descriptor
|
||||
// (*C.xce_job_submit_t: base jobid, array_size, one shared job_submit_data_t
|
||||
// spec — the same JobSubmit shape apiserver_push_jobarray() sends,
|
||||
|
|
@ -69,6 +78,7 @@ func cJobArrayToModels(jobArray *C.xce_job_submit_t) []*models.JobInfo {
|
|||
execHome: goString(si.exechome),
|
||||
execCWD: goString(si.execcwd),
|
||||
execUser: goString(si.execuser),
|
||||
exdata: goString(si.exdata),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -110,6 +120,7 @@ func cJobToModel(job *C.xce_job_t) *models.JobInfo {
|
|||
execHome: goString(si.exechome),
|
||||
execCWD: goString(si.execcwd),
|
||||
execUser: goString(si.execuser),
|
||||
exdata: goString(si.exdata),
|
||||
}
|
||||
}
|
||||
r.submit = rd
|
||||
|
|
@ -321,20 +332,16 @@ func cClusterConfToClusterInfo(conf *C.cluster_config_t) *models.ClusterInfo {
|
|||
// modelJobToSubmit converts a Go JobInfo to C xce_job_submit_t heap structs.
|
||||
// The caller must free all returned C memory by calling freeJobSubmit.
|
||||
func modelJobToSubmit(job *models.JobInfo) (*C.xce_job_submit_t, *C.job_submit_data_t, *C.job_startinfo_t) {
|
||||
rsi := submitStartInfo(job)
|
||||
si := (*C.job_startinfo_t)(C.calloc(1, C.sizeof_job_startinfo_t))
|
||||
si.cmdline = C.CString(job.Command)
|
||||
if execUser := resolveExecUser(job.ExecUser, job.SubmitUser); execUser != "" {
|
||||
si.execuser = C.CString(execUser)
|
||||
}
|
||||
if job.LaunchType != "" {
|
||||
si.launch_type = C.CString(job.LaunchType)
|
||||
}
|
||||
if job.PreExecCmd != "" {
|
||||
si.job_pre_exec = C.CString(job.PreExecCmd)
|
||||
}
|
||||
if job.PostExecCmd != "" {
|
||||
si.job_post_exec = C.CString(job.PostExecCmd)
|
||||
}
|
||||
si.cmdline = C.CString(rsi.cmdline)
|
||||
si.execuser = cStringOrNil(rsi.execUser)
|
||||
si.launch_type = cStringOrNil(rsi.launchType)
|
||||
si.job_pre_exec = cStringOrNil(rsi.preExec)
|
||||
si.job_post_exec = cStringOrNil(rsi.postExec)
|
||||
si.exechome = cStringOrNil(rsi.execHome)
|
||||
si.execcwd = cStringOrNil(rsi.execCWD)
|
||||
si.exdata = cStringOrNil(rsi.exdata)
|
||||
|
||||
sd := (*C.job_submit_data_t)(C.calloc(1, C.sizeof_job_submit_data_t))
|
||||
sd.queue = C.CString(job.QueueName)
|
||||
|
|
@ -381,6 +388,9 @@ func freeJobSubmit(js *C.xce_job_submit_t, sd *C.job_submit_data_t, si *C.job_st
|
|||
C.free(unsafe.Pointer(si.launch_type))
|
||||
C.free(unsafe.Pointer(si.job_pre_exec))
|
||||
C.free(unsafe.Pointer(si.job_post_exec))
|
||||
C.free(unsafe.Pointer(si.exechome))
|
||||
C.free(unsafe.Pointer(si.execcwd))
|
||||
C.free(unsafe.Pointer(si.exdata))
|
||||
C.free(unsafe.Pointer(si))
|
||||
}
|
||||
if sd != nil {
|
||||
|
|
|
|||
|
|
@ -236,6 +236,55 @@ type rawStartInfo struct {
|
|||
execHome string
|
||||
execCWD string
|
||||
execUser string
|
||||
exdata string
|
||||
}
|
||||
|
||||
// submitStartInfo maps a Go JobInfo to the job_startinfo_t fields a
|
||||
// submission carries to the controller (and from there, unchanged, to the
|
||||
// worker's startjob request). modelJobToSubmit (convert.go) is the CGO
|
||||
// adapter that copies the result into the C struct.
|
||||
//
|
||||
// bug-1936: exdata (env, limits, transfers, interactive mode, built by the
|
||||
// submit handler into ExtendData), the exec cwd and the exec home were
|
||||
// never set, so none of them reached the worker.
|
||||
func submitStartInfo(job *models.JobInfo) rawStartInfo {
|
||||
execCWD := job.ExecCWD
|
||||
if execCWD == "" {
|
||||
execCWD = job.CWD
|
||||
}
|
||||
return rawStartInfo{
|
||||
cmdline: job.Command,
|
||||
launchType: job.LaunchType,
|
||||
preExec: job.PreExecCmd,
|
||||
postExec: job.PostExecCmd,
|
||||
execHome: job.ExecHome,
|
||||
execCWD: execCWD,
|
||||
execUser: resolveExecUser(job.ExecUser, job.SubmitUser),
|
||||
exdata: job.ExtendData.String,
|
||||
}
|
||||
}
|
||||
|
||||
// maxExdataLen is the longest exdata the worker can take: xcecd hands it to
|
||||
// xcecd_worker in job_assignment_t.exdata[EXDATA_LENGTH] (16 KiB,
|
||||
// src/worker/include/uv_ipc.h), NUL included.
|
||||
const maxExdataLen = 16*1024 - 1
|
||||
|
||||
// ValidateExdata checks a job's extended data before it is submitted: the
|
||||
// worker's parse_exdata() fails the job at launch on anything but a JSON
|
||||
// object, and a longer string than maxExdataLen would be cut short on its
|
||||
// way to the job process.
|
||||
func ValidateExdata(exdata string) error {
|
||||
if exdata == "" {
|
||||
return nil
|
||||
}
|
||||
if len(exdata) > maxExdataLen {
|
||||
return fmt.Errorf("extended job data is %d bytes, more than the limit of %d", len(exdata), maxExdataLen)
|
||||
}
|
||||
var obj map[string]interface{}
|
||||
if err := json.Unmarshal([]byte(exdata), &obj); err != nil {
|
||||
return errors.New("extended job data must be a JSON object")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// rawSubmitData mirrors the C job_submit_data_t fields used by the Go model.
|
||||
|
|
@ -358,7 +407,9 @@ func rawJobToModel(r rawJobData) *models.JobInfo {
|
|||
info.PostExecCmd = si.postExec
|
||||
info.ExecHome = si.execHome
|
||||
info.ExecCWD = si.execCWD
|
||||
info.CWD = si.execCWD // the submit cwd is sent as the exec cwd (bug-1936)
|
||||
info.ExecUser = si.execUser
|
||||
info.ExtendData = models.Jsonb{String: si.exdata}
|
||||
}
|
||||
}
|
||||
info.SubmitTime = r.subTime
|
||||
|
|
|
|||
|
|
@ -1251,3 +1251,99 @@ func TestJobInfoExposesRunUsageColumns(t *testing.T) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
// bug-1936: a REST submission's cwd, exec home and extended data (env,
|
||||
// limits incl. the -W run limit, transfers, interactive mode) must reach
|
||||
// the job_startinfo_t modelJobToSubmit sends to the controller, which
|
||||
// passes it unchanged to the worker's startjob request.
|
||||
func TestSubmitStartInfo_CarriesExdataCwdAndHome(t *testing.T) {
|
||||
exdata := `{"limits":{"run_limit":60},"sub_env_vars":{"A":"1"}}`
|
||||
job := &models.JobInfo{
|
||||
Command: "sleep 120",
|
||||
SubmitUser: "alice",
|
||||
CWD: "/tmp/work",
|
||||
ExecHome: "/home/alice",
|
||||
ExtendData: models.Jsonb{String: exdata},
|
||||
}
|
||||
|
||||
si := submitStartInfo(job)
|
||||
|
||||
if si.exdata != exdata {
|
||||
t.Errorf("exdata: got %q, want %q", si.exdata, exdata)
|
||||
}
|
||||
if si.execCWD != "/tmp/work" {
|
||||
t.Errorf("execCWD: got %q, want %q", si.execCWD, "/tmp/work")
|
||||
}
|
||||
if si.execHome != "/home/alice" {
|
||||
t.Errorf("execHome: got %q, want %q", si.execHome, "/home/alice")
|
||||
}
|
||||
if si.cmdline != "sleep 120" || si.execUser != "alice" {
|
||||
t.Errorf("cmdline/execUser: got %q/%q, want %q/%q", si.cmdline, si.execUser, "sleep 120", "alice")
|
||||
}
|
||||
}
|
||||
|
||||
// bug-1936: exdata now reaches the worker, whose parse_exdata() fails the
|
||||
// job on anything but a JSON object and whose IPC copy holds at most
|
||||
// EXDATA_LENGTH-1 bytes; such a submission is rejected up front instead.
|
||||
func TestValidateExdata(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
exdata string
|
||||
wantErr bool
|
||||
}{
|
||||
{"empty", "", false},
|
||||
{"object", `{"limits":{"run_limit":60}}`, false},
|
||||
{"not json", "abc", true},
|
||||
{"array", `[{"op":">"}]`, true},
|
||||
{"string", `"abc"`, true},
|
||||
{"at limit", `{"k":"` + strings.Repeat("x", maxExdataLen-8) + `"}`, false},
|
||||
{"over limit", `{"k":"` + strings.Repeat("x", maxExdataLen-7) + `"}`, true},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
err := ValidateExdata(tt.exdata)
|
||||
if (err != nil) != tt.wantErr {
|
||||
t.Errorf("ValidateExdata(len %d) err = %v, wantErr %v", len(tt.exdata), err, tt.wantErr)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// bug-1936: an explicit exec cwd wins over the submit cwd.
|
||||
func TestSubmitStartInfo_PrefersExecCWD(t *testing.T) {
|
||||
si := submitStartInfo(&models.JobInfo{CWD: "/submit", ExecCWD: "/exec"})
|
||||
if si.execCWD != "/exec" {
|
||||
t.Errorf("execCWD: got %q, want %q", si.execCWD, "/exec")
|
||||
}
|
||||
}
|
||||
|
||||
// bug-1936: the start info the controller pushes back (single job and the
|
||||
// compact job-array descriptor) restores the cwd and extended data into the
|
||||
// store's JobInfo, so the apiserver's view keeps what was submitted.
|
||||
func TestRawJobArrayToModels_RestoresExdataAndCwd(t *testing.T) {
|
||||
exdata := `{"limits":{"run_limit":60}}`
|
||||
submit := &rawSubmitData{
|
||||
queue: "normal",
|
||||
startInfo: &rawStartInfo{
|
||||
cmdline: "sleep 1",
|
||||
execCWD: "/tmp/work",
|
||||
execHome: "/home/alice",
|
||||
exdata: exdata,
|
||||
},
|
||||
}
|
||||
jobs := rawJobArrayToModels(uint64(tagJobArray)<<tagShift|1, 2, submit)
|
||||
if len(jobs) != 2 {
|
||||
t.Fatalf("elements: got %d, want 2", len(jobs))
|
||||
}
|
||||
for i, j := range jobs {
|
||||
if j.ExtendData.String != exdata {
|
||||
t.Errorf("element %d ExtendData: got %q, want %q", i, j.ExtendData.String, exdata)
|
||||
}
|
||||
if j.CWD != "/tmp/work" || j.ExecCWD != "/tmp/work" {
|
||||
t.Errorf("element %d CWD/ExecCWD: got %q/%q, want /tmp/work", i, j.CWD, j.ExecCWD)
|
||||
}
|
||||
if j.ExecHome != "/home/alice" {
|
||||
t.Errorf("element %d ExecHome: got %q, want /home/alice", i, j.ExecHome)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1920,7 +1920,9 @@ func processJobRequest(ctx *gin.Context, req *models.JobSubmitRequest, defaultVa
|
|||
extendData[i] = entry
|
||||
}
|
||||
}
|
||||
extendDataMap["transfer_files"] = extendData
|
||||
if len(extendData) > 0 {
|
||||
extendDataMap["transfer_files"] = extendData
|
||||
}
|
||||
}
|
||||
|
||||
// handle env into json format
|
||||
|
|
@ -1990,6 +1992,12 @@ func processJobRequest(ctx *gin.Context, req *models.JobSubmitRequest, defaultVa
|
|||
req.ExtendData = string(updatedExtendDataBytes)
|
||||
}
|
||||
|
||||
// bug-1936: ExtendData is the job's exdata at the worker; reject what
|
||||
// the worker could not parse instead of failing the job at launch.
|
||||
if err := backend.ValidateExdata(req.ExtendData); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1519,3 +1519,60 @@ func TestJobHandler_RestartNamedJob(t *testing.T) {
|
|||
w.Code, len(f.submitted)-before, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// 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,
|
||||
// sub_env_vars as a string map); modelJobToSubmit copies them into the
|
||||
// start info.
|
||||
func TestJobHandler_SubmitCarriesCwdEnvLimitsInExdata(t *testing.T) {
|
||||
f := newHandlerFixture(t)
|
||||
f.seedQueue("normal")
|
||||
|
||||
w := f.submit(map[string]interface{}{
|
||||
"command": "sleep 120", "queue_name": "normal", "cwd": "/tmp/work",
|
||||
"env": []string{"A=1"}, "run_limit": "1", "mem_limit": "1 GB",
|
||||
})
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200; body: %s", w.Code, w.Body.String())
|
||||
}
|
||||
subs := f.submissions()
|
||||
if len(subs) != 1 {
|
||||
t.Fatalf("submitJob calls = %d, want 1", len(subs))
|
||||
}
|
||||
job := subs[0].Job
|
||||
if job.CWD != "/tmp/work" {
|
||||
t.Errorf("CWD = %q, want /tmp/work", job.CWD)
|
||||
}
|
||||
var ex struct {
|
||||
Env map[string]string `json:"sub_env_vars"`
|
||||
Limits map[string]float64 `json:"limits"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(job.ExtendData.String), &ex); err != nil {
|
||||
t.Fatalf("ExtendData %q is not a JSON object: %v", job.ExtendData.String, err)
|
||||
}
|
||||
if ex.Env["A"] != "1" {
|
||||
t.Errorf("sub_env_vars = %v, want A=1", ex.Env)
|
||||
}
|
||||
if ex.Limits["run_limit"] != 60 || ex.Limits["rss_limit"] != 1024 {
|
||||
t.Errorf("limits = %v, want run_limit 60 (s) and rss_limit 1024 (MB)", ex.Limits)
|
||||
}
|
||||
}
|
||||
|
||||
// bug-1936: extend_data the worker cannot parse (not a JSON object) would
|
||||
// fail the job at launch; the submission is rejected instead.
|
||||
func TestJobHandler_SubmitRejectsNonObjectExtendData(t *testing.T) {
|
||||
f := newHandlerFixture(t)
|
||||
f.seedQueue("normal")
|
||||
|
||||
for _, ex := range []string{"abc", `[1,2]`} {
|
||||
before := len(f.submissions())
|
||||
w := f.submit(map[string]interface{}{
|
||||
"command": "sleep 1", "queue_name": "normal", "extend_data": ex,
|
||||
})
|
||||
if w.Code == http.StatusOK || len(f.submissions()) != before {
|
||||
t.Errorf("extend_data %q: status %d, %d submits; want an error and none; body %s",
|
||||
ex, w.Code, len(f.submissions())-before, w.Body.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue