diff --git a/src/frontend/apiserver/controllers/jobs.go b/src/frontend/apiserver/controllers/jobs.go index 22b9fb6c..ba80eb0f 100644 --- a/src/frontend/apiserver/controllers/jobs.go +++ b/src/frontend/apiserver/controllers/jobs.go @@ -2048,6 +2048,14 @@ func processJobRequest(ctx *gin.Context, req *models.JobSubmitRequest, defaultVa extendDataMap["interactive"] = req.Interactive } + // attach=1: a TTY job whose session ends with its command (xcecli -I + // without --browse). The worker uses it to reap ttyd when no client ever + // connects -- without it an unattached job would sit in RUN forever, + // holding its ttyd, port and per-job worker. + if req.Attach != 0 { + extendDataMap["attach"] = req.Attach + } + if len(extendDataMap) > 0 { updatedExtendDataBytes, err := json.Marshal(extendDataMap) if err != nil { diff --git a/src/frontend/apiserver/controllers/jobs_handler_test.go b/src/frontend/apiserver/controllers/jobs_handler_test.go index 3dc1811a..1e55408e 100644 --- a/src/frontend/apiserver/controllers/jobs_handler_test.go +++ b/src/frontend/apiserver/controllers/jobs_handler_test.go @@ -428,6 +428,56 @@ func TestJobHandler_SmokeSubmitErrorIs500(t *testing.T) { } } +// TestJobHandler_SubmitInteractiveAttachExdata checks the submit handler +// packs interactive/attach into the job's ExtendData: attach=1 only for the +// terminal-attach TTY kind (the worker reaps ttyd when no client ever +// connects), never for --browse or other interactive modes. +func TestJobHandler_SubmitInteractiveAttachExdata(t *testing.T) { + cases := []struct { + name string + req map[string]interface{} + wantAttach bool + }{ + {"tty attach", map[string]interface{}{ + "job_name": "a", "command": "sleep 1", "interactive": 1, "attach": 1, + }, true}, + {"tty web view (no attach)", map[string]interface{}{ + "job_name": "a", "command": "sleep 1", "interactive": 1, + }, false}, + {"xjob with attach ignored worker-side but passed", map[string]interface{}{ + "job_name": "a", "command": "sleep 1", "interactive": 2, "attach": 1, + }, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := newHandlerFixture(t) + f.seedQueue("normal") + + w := f.submit(tc.req) + 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)) + } + var ex map[string]interface{} + if raw := subs[0].Job.ExtendData.String; raw != "" { + if err := json.Unmarshal([]byte(raw), &ex); err != nil { + t.Fatalf("ExtendData %q is not JSON: %v", raw, err) + } + } + if got, ok := ex["interactive"]; !ok || int(got.(float64)) != tc.req["interactive"].(int) { + t.Errorf("ExtendData interactive = %v (%v), want %d", got, ok, tc.req["interactive"]) + } + _, hasAttach := ex["attach"] + if hasAttach != tc.wantAttach { + t.Errorf("ExtendData attach present = %v, want %v (exdata: %s)", hasAttach, tc.wantAttach, subs[0].Job.ExtendData.String) + } + }) + } +} + // TestJobHandler_SmokeControlByID controls one seeded ordinary job by ID // and checks the dispatch is recorded once with its unique_job_id. func TestJobHandler_SmokeControlByID(t *testing.T) { diff --git a/src/frontend/apiserver/models/jobs.go b/src/frontend/apiserver/models/jobs.go index 8b743d78..b84223af 100644 --- a/src/frontend/apiserver/models/jobs.go +++ b/src/frontend/apiserver/models/jobs.go @@ -669,6 +669,11 @@ type JobSubmitRequest struct { Exclusive bool `json:"exclusive,omitempty"` Rerunable bool `json:"rerunable,omitempty"` Interactive int `json:"interactive,omitempty"` + // Attach marks an interactive TTY job as the terminal-attach kind + // (xcecli -I without --browse): the session ends with the command, so + // the worker may reap the ttyd when no client ever connects. 0 keeps + // the web-viewed semantics. + Attach int `json:"attach,omitempty"` WarningTimePeriod int `json:"warning_time_period,omitempty"` WarningAction string `json:"warning_action,omitempty"` DataSpecFile string `json:"data_spec_file,omitempty"` diff --git a/src/frontend/xcecli/cmd/jobs/submit.go b/src/frontend/xcecli/cmd/jobs/submit.go index a1577585..ce9f2178 100755 --- a/src/frontend/xcecli/cmd/jobs/submit.go +++ b/src/frontend/xcecli/cmd/jobs/submit.go @@ -253,6 +253,12 @@ func (js *JobSubmitter) BuildJobRequest() (*models.JobSubmitRequest, error) { if js.interactive { if js.useTTY { jobReq.Interactive = 1 // TTY mode + // attach jobs end with their command, web-viewed ones keep the + // session alive for a later browser; the worker needs the + // distinction to reap a ttyd no client ever connected to + if !js.useBrowse { + jobReq.Attach = 1 + } } else if js.useXJob { jobReq.Interactive = 2 // XJob mode } else { @@ -520,7 +526,11 @@ func (js *JobSubmitter) RunJobsSubmit() error { // user asked for the web view. --browse opens the ttyd web URL // in the browser instead of attaching. if !js.useBrowse { - return runAttachSession(serverInfo.URL, serverInfo.Token, jobID) + err := runAttachSession(serverInfo.URL, serverInfo.Token, jobID) + if err != nil { + fmt.Println("note: the job will be cleaned up by the worker if no client attaches within 60 s") + } + return err } if err := openBrowser(url); err != nil { errMsg := request.ExtractErrorMessage(err) diff --git a/src/frontend/xcecli/cmd/jobs/submit_test.go b/src/frontend/xcecli/cmd/jobs/submit_test.go index e951546c..3ccf2e74 100644 --- a/src/frontend/xcecli/cmd/jobs/submit_test.go +++ b/src/frontend/xcecli/cmd/jobs/submit_test.go @@ -133,6 +133,47 @@ func TestSubmitHelpShowsArrayForms(t *testing.T) { } } +// Attach marks the terminal-attach kind of TTY job so the worker may reap +// ttyd when no client ever connects; --browse keeps the web-viewed +// semantics and must not set it. +func TestBuildJobRequest_AttachFlagMatrix(t *testing.T) { + cases := []struct { + name string + interactive bool + useTTY bool + useBrowse bool + useXJob bool + wantInter int + wantAttach int + }{ + {"batch job", false, false, false, false, 0, 0}, + {"tty attach (bare -I)", true, true, false, false, 1, 1}, + {"tty browse", true, true, true, false, 1, 0}, + {"xjob", true, false, false, true, 2, 0}, + {"display (interactive=3)", true, false, false, false, 3, 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + js := NewJobSubmitter(nil) + js.command = "sleep 1" + js.interactive = tc.interactive + js.useTTY = tc.useTTY + js.useBrowse = tc.useBrowse + js.useXJob = tc.useXJob + req, err := js.BuildJobRequest() + if err != nil { + t.Fatalf("BuildJobRequest: %v", err) + } + if req.Interactive != tc.wantInter { + t.Errorf("Interactive = %d, want %d", req.Interactive, tc.wantInter) + } + if req.Attach != tc.wantAttach { + t.Errorf("Attach = %d, want %d", req.Attach, tc.wantAttach) + } + }) + } +} + // jobStateServer answers the per-job status query with the states fed in // sequence, holding the last one forever; every request's arrival time is // recorded so tests can assert the poll cadence. @@ -167,9 +208,9 @@ func newJobStateServer(t *testing.T, states ...string) *jobStateServer { _ = json.NewEncoder(w).Encode(models.JobsResponse{ Code: 0, Data: struct { - Count int `json:"count"` + Count int `json:"count"` Rows []models.QueryJob `json:"rows"` - Page_token string `json:"page_token"` + Page_token string `json:"page_token"` }{Rows: []models.QueryJob{{JobStatus: held, InteractiveJobURL: "http://h:9000"}}}, }) })) @@ -250,9 +291,9 @@ func TestWaitForInteractiveURL_RunRaceUsesFastInterval(t *testing.T) { _ = json.NewEncoder(w).Encode(models.JobsResponse{ Code: 0, Data: struct { - Count int `json:"count"` + Count int `json:"count"` Rows []models.QueryJob `json:"rows"` - Page_token string `json:"page_token"` + Page_token string `json:"page_token"` }{Rows: []models.QueryJob{row}}, }) })) diff --git a/src/frontend/xcecli/models/jobs.go b/src/frontend/xcecli/models/jobs.go index 61187431..53dfb43d 100755 --- a/src/frontend/xcecli/models/jobs.go +++ b/src/frontend/xcecli/models/jobs.go @@ -343,6 +343,11 @@ type JobSubmitRequest struct { Exclusive bool `json:"exclusive,omitempty"` Rerunable bool `json:"rerunable,omitempty"` Interactive int `json:"interactive,omitempty"` + // Attach marks an interactive TTY job as the terminal-attach kind (xcecli + // -I without --browse): its session ends with the command, so the worker + // may reap the ttyd when no client ever connects. 0 (default) keeps the + // web-viewed semantics. + Attach int `json:"attach,omitempty"` WarningTimePeriod int `json:"warning_time_period,omitempty"` WarningAction string `json:"warning_action,omitempty"` DataSpecFile string `json:"data_spec_file,omitempty"` diff --git a/src/worker/include/xcecd_exdata.h b/src/worker/include/xcecd_exdata.h index 0197fadb..8c9a9f5c 100644 --- a/src/worker/include/xcecd_exdata.h +++ b/src/worker/include/xcecd_exdata.h @@ -24,6 +24,10 @@ typedef struct { typedef struct { transfer_file_t* ops; int interactive_mode; + /* attach=1 (exdata key "attach"): TTY job whose session ends with its + * command (xcecli attach path). The worker may reap the ttyd when no + * client ever connects; web-viewed jobs (0) keep it for a later browser. */ + int attach_mode; int count; } transfer_info_t; diff --git a/src/worker/src/xcecd_exdata.c b/src/worker/src/xcecd_exdata.c index 9a1948c8..8e559a28 100644 --- a/src/worker/src/xcecd_exdata.c +++ b/src/worker/src/xcecd_exdata.c @@ -236,6 +236,22 @@ bool parse_exdata(const char* exdata, env_var_t** env_vars, XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Parsed interactive mode: %d", transfer_infos->interactive_mode); } + // --- Handle 'attach' --- + else if (strcmp(key, "attach") == 0) { + if (cJSON_IsString(item) && item->valuestring != NULL) { + transfer_infos->attach_mode = atoi(item->valuestring); + } else if (cJSON_IsNumber(item)) { + transfer_infos->attach_mode = item->valueint; + } else { + XCECD_WORKER_LOG( + XCE_LOG_ERROR, + "Error: 'attach' is neither a string nor a number."); + success = false; + continue; + } + XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Parsed attach mode: %d", + transfer_infos->attach_mode); + } // --- Skip 'limits' --- else if (strcmp(key, "limits") == 0) { // limits will be handled in parse_limit diff --git a/src/worker/src/xcecd_worker.c b/src/worker/src/xcecd_worker.c index cd4c347c..e5fe4982 100644 --- a/src/worker/src/xcecd_worker.c +++ b/src/worker/src/xcecd_worker.c @@ -53,6 +53,22 @@ static job_signal_seq_t graceful_seq; static bool graceful_seq_ready = false; static int job_rusage_collect_interval = 10000; // 10 seconds +/* Attach TTY watchdog: ttyd only starts the job command on the first client + * connection, so with no client ever arriving the command never runs, ttyd + * never exits and the job would sit in RUN forever holding its ttyd, port + * and per-job worker. For attach jobs (exdata "attach":1) the timer below + * reaps ttyd via tty_stop_attach_session, which lands us in the regular + * process_exit_callback path (JOB_EXITED, status reason, post-exec, exit). */ +#define TTY_ATTACH_WATCHDOG_MS 1000 /* poll cadence */ +#define TTY_ATTACH_NO_CLIENT_MS 60000 /* no client ever connected */ +#define TTY_ATTACH_CHILD_GONE_MS 5000 /* client came, command ended */ +static uv_timer_t tty_attach_timer; +static bool tty_attach_timer_ready = false; +static int64_t tty_attach_start_ms = 0; +static bool tty_attach_saw_child = false; +static int64_t tty_attach_child_gone_ms = 0; +static const char* tty_reap_reason = NULL; + // process manager will share the same event loop with ipc uv_loop_t* uv_loop = NULL; @@ -63,6 +79,9 @@ static void xjob_stop_vnc(); static void tty_stop_attach_session(); static bool check_process_status(pid_t target_pid, int expected_signal, int timeout_seconds); +static pid_t ttyd_real_pid(pid_t monitored_pid); +static void tty_attach_watchdog_start(); +static void tty_attach_watchdog_stop(); static void free_xce_job_rusages(xce_job_rusages_t* rusages) { @@ -477,6 +496,8 @@ static void process_exit_callback(int pid, int exit_status, int term_signal, job_signal_seq_cancel(&graceful_seq); } + tty_attach_watchdog_stop(); + check_and_reconnect(); stop_rusage_collection(); @@ -507,7 +528,10 @@ static void process_exit_callback(int pid, int exit_status, int term_signal, XCECD_WORKER_LOG(XCE_LOG_INFO, "Child %d killed by signal %d\n", pid, term_signal); - ipc_worker_send_status(&job_runtime_info, "Killed"); + /* the watchdog's SIGKILL reaps ttyd through this callback: report + * why instead of a bare "Killed" */ + ipc_worker_send_status(&job_runtime_info, + tty_reap_reason ? tty_reap_reason : "Killed"); return_code = exit_status; } @@ -730,25 +754,258 @@ static void xjob_stop_vnc() * is reaped; tty_stop_attach_session is only called for the attach path. */ static void tty_stop_attach_session() { - pid_t ttyd_pid = uv_proc_get_pid(&uv_proc_manager); + pid_t monitored_pid = uv_proc_get_pid(&uv_proc_manager); - if (ttyd_pid <= 0) { + if (monitored_pid <= 0) { return; } char jobidStr[64]; fmt_jobid(jobidStr, sizeof(jobidStr), jobid); XCECD_WORKER_LOG(XCE_LOG_DEBUG, "Terminating ttyd %d of attach job %s", - ttyd_pid, jobidStr); + ttyd_real_pid(monitored_pid), jobidStr); - signal_children_recursive(SIGKILL, ttyd_pid); - if (kill(ttyd_pid, SIGKILL) != 0 && errno != ESRCH) { + /* kill from the monitored "/bin/sh -c" down: covers both topologies + * (sh exec'd ttyd -- the monitored pid is ttyd; sh forked it -- the + * real ttyd hangs off the wrapper) */ + signal_children_recursive(SIGKILL, monitored_pid); + if (kill(monitored_pid, SIGKILL) != 0 && errno != ESRCH) { XCECD_WORKER_LOG(XCE_LOG_WARN, - "Failed to kill ttyd %d of job %s: %s", ttyd_pid, + "Failed to kill ttyd %d of job %s: %s", monitored_pid, jobidStr, strerror(errno)); } } +/* First direct child of `parent`, from a /proc scan (same read as + * signal_children_recursive); 0 when it has none. */ +static pid_t first_child_of(pid_t parent) +{ + DIR* dir; + struct dirent* entry; + char path[256]; + char buf[MAXLINELEN]; + FILE* stat_file; + pid_t child = 0; + + if (parent <= 0 || (dir = opendir("/proc")) == NULL) { + return 0; + } + + while ((entry = readdir(dir)) != NULL && child == 0) { + pid_t pid; + + if (sscanf(entry->d_name, "%d", &pid) != 1) + continue; + + if (pid == parent) + continue; + + snprintf(path, sizeof(path), "/proc/%d/stat", pid); + stat_file = fopen(path, "r"); + if (!stat_file) + continue; + + if (fgets(buf, sizeof(buf), stat_file)) { + int parsed_pid; + char comm[256]; + char state; + pid_t ppid; + /* "pid (comm) state ppid": tolerant of spaces in comm */ + if (sscanf(buf, "%d (%255[^)]) %c %d", &parsed_pid, comm, &state, + &ppid) + == 4 + && ppid == parent) { + child = pid; + } + } + fclose(stat_file); + } + + closedir(dir); + return child; +} + +/* The worker monitors "/bin/sh -c ttyd ...": sh either execs ttyd (the + * monitored pid IS ttyd) or forks it (ttyd hangs off the wrapper -- seen on + * the field). Resolve the real ttyd pid so the watchdog judges client + * activity by the ttyd process's children; against the wrapper, ttyd itself + * would count as "the command has started" and the no-client reap would + * never fire. */ +static pid_t ttyd_real_pid(pid_t monitored_pid) +{ + char path[64]; + char comm[64]; + FILE* fp; + char* line; + + pid_t pid = monitored_pid; + + for (int depth = 0; pid > 0 && depth < 3; depth++) { + snprintf(path, sizeof(path), "/proc/%d/comm", (int)pid); + fp = fopen(path, "r"); + if (!fp) { + return 0; /* gone or unreadable: ttyd not (yet) visible */ + } + line = fgets(comm, sizeof(comm), fp); + fclose(fp); + if (!line) { + return 0; + } + comm[strcspn(comm, "\n")] = '\0'; + + if (strcmp(comm, "ttyd") == 0) { + return pid; + } + + pid = first_child_of(pid); /* descend through the sh wrapper */ + } + + return 0; +} + +/* True when the real ttyd has forked its child command: ttyd only does that + * once the first websocket client connects, so a child is both proof of a + * client and of the command having started. */ +static bool ttyd_has_child(pid_t ttyd_pid) +{ + DIR* dir; + struct dirent* entry; + char path[256]; + char buf[MAXLINELEN]; + FILE* stat_file; + pid_t ppid; + + if (ttyd_pid <= 0) { + return false; + } + + if ((dir = opendir("/proc")) == NULL) { + /* unreadable /proc must not reap a live session */ + return true; + } + + while ((entry = readdir(dir)) != NULL) { + pid_t pid; + + if (sscanf(entry->d_name, "%d", &pid) != 1) + continue; + + if (pid == ttyd_pid) + continue; + + snprintf(path, sizeof(path), "/proc/%d/stat", pid); + stat_file = fopen(path, "r"); + if (!stat_file) + continue; + + if (fgets(buf, sizeof(buf), stat_file)) { + int parsed_pid; + char comm[256]; + char state; + /* "pid (comm) state ppid": tolerant of spaces in comm */ + if (sscanf(buf, "%d (%255[^)]) %c %d", &parsed_pid, comm, &state, + &ppid) + == 4 + && ppid == ttyd_pid) { + fclose(stat_file); + closedir(dir); + return true; + } + } + fclose(stat_file); + } + + closedir(dir); + return false; +} + +static void tty_attach_watchdog_callback(uv_timer_t* timer) +{ + (void)timer; + + if (worker_exit_flag || TTY != mydata.transfer_info->interactive_mode + || 1 != mydata.transfer_info->attach_mode) { + return; + } + + pid_t monitored_pid = uv_proc_get_pid(&uv_proc_manager); + if (monitored_pid <= 0) { + return; /* process_exit_callback owns the job now */ + } + + /* descend "/bin/sh -c" (it may have exec'd or forked ttyd) */ + pid_t ttyd_pid = ttyd_real_pid(monitored_pid); + if (ttyd_pid <= 0) { + /* ttyd not visible yet (sh still starting) or already gone (the + * process-exit path owns the job): hold off, do not reap */ + return; + } + + int64_t now_ms = uv_now(uv_loop); + + if (ttyd_has_child(ttyd_pid)) { + if (!tty_attach_saw_child) { + tty_attach_saw_child = true; + char jobidStr[64]; + fmt_jobid(jobidStr, sizeof(jobidStr), jobid); + XCECD_WORKER_LOG(XCE_LOG_DEBUG, + "Attach job %s: command started (client " + "connected)", + jobidStr); + } + tty_attach_child_gone_ms = 0; + return; + } + + char jobidStr[64]; + fmt_jobid(jobidStr, sizeof(jobidStr), jobid); + + if (tty_attach_saw_child) { + /* the command ran and is gone; give ttyd a moment to exit on its + * own (-Q), then reap -- its teardown cannot be relied on with a + * relay client still connected */ + if (tty_attach_child_gone_ms == 0) { + tty_attach_child_gone_ms = now_ms; + } else if (now_ms - tty_attach_child_gone_ms + >= TTY_ATTACH_CHILD_GONE_MS) { + XCECD_WORKER_LOG( + XCE_LOG_NOTICE, + "Attach job %s: command exited but ttyd %d lingers, " + "reaping it", + jobidStr, ttyd_pid); + tty_reap_reason = "Command exited; ttyd did not stop on its own"; + tty_stop_attach_session(); + } + return; + } + + if (now_ms - tty_attach_start_ms >= TTY_ATTACH_NO_CLIENT_MS) { + XCECD_WORKER_LOG(XCE_LOG_NOTICE, + "Attach job %s: no client attached within %d s, " + "reaping ttyd %d", + jobidStr, (int)(TTY_ATTACH_NO_CLIENT_MS / 1000), + ttyd_pid); + tty_reap_reason = "No client attached within 60 s"; + tty_stop_attach_session(); + } +} + +static void tty_attach_watchdog_start() +{ + tty_attach_start_ms = uv_now(uv_loop); + tty_attach_saw_child = false; + tty_attach_child_gone_ms = 0; + uv_timer_start(&tty_attach_timer, tty_attach_watchdog_callback, + TTY_ATTACH_WATCHDOG_MS, TTY_ATTACH_WATCHDOG_MS); +} + +static void tty_attach_watchdog_stop() +{ + if (tty_attach_timer_ready) { + uv_timer_stop(&tty_attach_timer); + } +} + static void worker_job_callback(job_assignment_t* info) { char* job_cmd = NULL; @@ -938,6 +1195,15 @@ static void worker_job_callback(job_assignment_t* info) start_rusage_collection(); XCECD_WORKER_LOG(XCE_LOG_DEBUG3, "Starting job rusage collection..."); + tty_reap_reason = NULL; + + /* attach TTY jobs: reap ttyd if no client ever connects (the job + * command would otherwise never start -- ttyd is on-demand) */ + if (TTY == mydata.transfer_info->interactive_mode + && 1 == mydata.transfer_info->attach_mode) { + tty_attach_watchdog_start(); + } + if (mydata.transfer_info->interactive_mode > 0) { url = generate_url(port_info, mydata.transfer_info->interactive_mode); @@ -1611,6 +1877,10 @@ int main(int argc, char* argv[]) /* bug-1957: timer for graceful kill/stop sequences */ graceful_seq_ready = (job_signal_seq_init(&graceful_seq, uv_loop) == 0); + /* attach TTY watchdog (tty_attach_watchdog_callback) */ + uv_timer_init(uv_loop, &tty_attach_timer); + tty_attach_timer_ready = true; + do { ipc_worker_run(UV_RUN_NOWAIT);