xcecli: backoff poll cadence and suspend tolerance for -I URL wait
waitForInteractiveURL polled PEND every 5s, adding ~4.4s of dead time to every interactive submit (the job typically reaches RUN with its session URL ~0.6s after submit once the apiserver applies the controller's jobPidInfo push), and the RUN-but-no-URL-yet round waited another 2s. Front-load fast 500ms polls (10 rounds) so a normally dispatched job attaches within ~1s, then back off to 2s so a job stuck in a long queue costs the server a query every couple of seconds instead of a permanent 2 qps; the RUN race polls 500ms too. checkJobStatus also failed the wait outright on any state other than PEND/RUN: treat PSUSP like PEND (still queued), and USUSP/SSUSP/ TRANSIENT as keep-waiting states — the allocation is held, a resume (or failover settle) may follow, and tearing down the attach on a transient state is wrong. Cluster-verified on t1: submit-to-attach-attempt 0.56-1.08s across three runs (previously ~6s even after the jobPidInfo fix), full submit/run/finish lifecycle clean, daemons healthy. Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
lei.chen
parent
3545aa0d77
commit
6ce7bdbe11
2 changed files with 211 additions and 7 deletions
|
|
@ -576,7 +576,7 @@ func (js *JobSubmitter) checkJobStatus(apiClient *request.APIClient, token strin
|
|||
}
|
||||
for _, row := range resp.Data.Rows {
|
||||
switch row.JobStatus {
|
||||
case "PEND":
|
||||
case "PEND", "PSUSP":
|
||||
return row.JobStatus, nil
|
||||
case "RUN":
|
||||
// Print the dispatch line once: waitForInteractiveURL keeps
|
||||
|
|
@ -590,6 +590,17 @@ func (js *JobSubmitter) checkJobStatus(apiClient *request.APIClient, token strin
|
|||
}
|
||||
}
|
||||
return row.JobStatus, nil
|
||||
case "USUSP", "SSUSP":
|
||||
// A suspended job keeps its allocation and can be resumed
|
||||
// (`xce jobs resume`), so the session may still come up --
|
||||
// wait instead of failing the attach the way a terminal
|
||||
// state (DONE/EXIT) does.
|
||||
return row.JobStatus, nil
|
||||
case "TRANSIENT":
|
||||
// State uncertain after a controller failover; the job may
|
||||
// settle back to RUN. Keep waiting (the 24h cap still bounds
|
||||
// this) rather than tearing down the attach on a hiccup.
|
||||
return row.JobStatus, nil
|
||||
default:
|
||||
return "", fmt.Errorf("job %s is in state %s, please check the job status", jobID, row.JobStatus)
|
||||
}
|
||||
|
|
@ -622,26 +633,57 @@ func (js *JobSubmitter) checkExtendData(apiClient *request.APIClient, token stri
|
|||
return "", nil
|
||||
}
|
||||
|
||||
// Interactive-wait poll cadence. A dispatched job produces its session URL
|
||||
// well under a second after submit (scheduler round ~0.5s, start report
|
||||
// right behind it), so pendingInterval fronts fast polls to make attach
|
||||
// snappy; pendingBackoffInterval takes over after pendingFastPolls so a job
|
||||
// stuck in a long queue (closed window, no free slots) costs the server one
|
||||
// query every couple of seconds instead of a permanent 2 qps. runInterval
|
||||
// covers the RUN-but-no-URL-yet race, which the start-report push (applied
|
||||
// by the apiserver's OnJobPidInfo) normally closes in well under a second.
|
||||
const (
|
||||
pendingFastPolls = 10
|
||||
pendingFastInterval = 500 * time.Millisecond
|
||||
pendingBackoffInterval = 2 * time.Second
|
||||
runInterval = 500 * time.Millisecond
|
||||
)
|
||||
|
||||
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()
|
||||
|
||||
polls := 0
|
||||
for {
|
||||
select {
|
||||
case <-timeout:
|
||||
return "", fmt.Errorf("timeout waiting for interactive URL")
|
||||
case <-timer.C:
|
||||
polls++
|
||||
status, err := js.checkJobStatus(apiClient, token, jobID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if status == "PEND" {
|
||||
timer.Reset(5 * time.Second)
|
||||
switch status {
|
||||
case "PEND", "PSUSP", "TRANSIENT":
|
||||
// Not running yet (or state uncertain after a controller
|
||||
// failover): fast-poll the first few rounds so a normally
|
||||
// dispatched job attaches with no perceptible delay, then
|
||||
// back off while the wait drags on.
|
||||
if polls <= pendingFastPolls {
|
||||
timer.Reset(pendingFastInterval)
|
||||
} else {
|
||||
timer.Reset(pendingBackoffInterval)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if status == "RUN" {
|
||||
case "USUSP", "SSUSP":
|
||||
// Suspended after starting: the allocation is held and a
|
||||
// resume may follow any moment, but there is nothing to
|
||||
// poll eagerly -- poll at the backed-off cadence.
|
||||
timer.Reset(pendingBackoffInterval)
|
||||
continue
|
||||
case "RUN":
|
||||
url, err := js.checkExtendData(apiClient, token, jobID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
|
|
@ -649,12 +691,13 @@ func (js *JobSubmitter) waitForInteractiveURL(apiClient *request.APIClient, toke
|
|||
if url == "" {
|
||||
// running but the worker's start report (carrying the
|
||||
// URL) has not arrived yet -- poll again shortly
|
||||
timer.Reset(2 * time.Second)
|
||||
timer.Reset(runInterval)
|
||||
continue
|
||||
}
|
||||
return url, nil
|
||||
default:
|
||||
return "", fmt.Errorf("job %s is in state %s, please check the job status", jobID, status)
|
||||
}
|
||||
return "", fmt.Errorf("job %s is in state %s, please check the job status", jobID, status)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,7 +7,9 @@ import (
|
|||
"net/http/httptest"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
|
|
@ -130,3 +132,162 @@ func TestSubmitHelpShowsArrayForms(t *testing.T) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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.
|
||||
type jobStateServer struct {
|
||||
srv *httptest.Server
|
||||
states chan string
|
||||
mu sync.Mutex
|
||||
pollAts []time.Time
|
||||
// urlQueries counts requests carrying fields=interactive_job_url.
|
||||
urlQueries int
|
||||
}
|
||||
|
||||
func newJobStateServer(t *testing.T, states ...string) *jobStateServer {
|
||||
t.Helper()
|
||||
js := &jobStateServer{states: make(chan string, len(states))}
|
||||
for _, s := range states {
|
||||
js.states <- s
|
||||
}
|
||||
held := ""
|
||||
js.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
js.mu.Lock()
|
||||
js.pollAts = append(js.pollAts, time.Now())
|
||||
if r.URL.Query().Get("fields") == "interactive_job_url" {
|
||||
js.urlQueries++
|
||||
}
|
||||
js.mu.Unlock()
|
||||
select {
|
||||
case held = <-js.states:
|
||||
default:
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(models.JobsResponse{
|
||||
Code: 0,
|
||||
Data: struct {
|
||||
Count int `json:"count"`
|
||||
Rows []models.QueryJob `json:"rows"`
|
||||
Page_token string `json:"page_token"`
|
||||
}{Rows: []models.QueryJob{{JobStatus: held, InteractiveJobURL: "http://h:9000"}}},
|
||||
})
|
||||
}))
|
||||
t.Cleanup(js.srv.Close)
|
||||
return js
|
||||
}
|
||||
|
||||
func (js *jobStateServer) pollCount() int {
|
||||
js.mu.Lock()
|
||||
defer js.mu.Unlock()
|
||||
return len(js.pollAts)
|
||||
}
|
||||
|
||||
// waitForInteractiveURL fast-polls the PEND phase (500ms) so a normally
|
||||
// dispatched job attaches within ~a second, then backs off to 2s so a job
|
||||
// waiting on a long queue doesn't hold 2 qps against the server forever.
|
||||
func TestWaitForInteractiveURL_PendFastThenBackoff(t *testing.T) {
|
||||
if config.Log == nil {
|
||||
config.Log = zap.NewNop()
|
||||
t.Cleanup(func() { config.Log = nil })
|
||||
}
|
||||
// PEND through the fast window (11 polls at t=0..5.0s) plus one backoff
|
||||
// round, then RUN+URL: the total wait is 10*0.5 + 2 + ~0 = ~7s.
|
||||
states := make([]string, 11)
|
||||
for i := range states {
|
||||
states[i] = "PEND"
|
||||
}
|
||||
states = append(states, "RUN")
|
||||
js := newJobStateServer(t, states...)
|
||||
cli := NewJobSubmitter(nil)
|
||||
start := time.Now()
|
||||
url, err := cli.waitForInteractiveURL(request.NewAPIClient(js.srv.URL), "tok", "123")
|
||||
elapsed := time.Since(start)
|
||||
if err != nil {
|
||||
t.Fatalf("waitForInteractiveURL: %v", err)
|
||||
}
|
||||
if url != "http://h:9000" {
|
||||
t.Errorf("url = %q, want http://h:9000", url)
|
||||
}
|
||||
// ~7s expected; under 6s means the backoff never happened (all fast),
|
||||
// over 9s means the fast window was skipped (all 2s).
|
||||
if elapsed < 6*time.Second || elapsed > 9*time.Second {
|
||||
t.Errorf("elapsed = %v, want ~7s (10 fast intervals then a 2s backoff)", elapsed)
|
||||
}
|
||||
if n := js.pollCount(); n != 13 {
|
||||
t.Errorf("polls = %d, want 13 (12 status + 1 url query)", n)
|
||||
}
|
||||
}
|
||||
|
||||
// RUN-before-URL polls at the 500ms run cadence: the start-report push
|
||||
// normally closes the race in well under a second, so a 2s wait here would
|
||||
// add that much dead time to every interactive submit.
|
||||
func TestWaitForInteractiveURL_RunRaceUsesFastInterval(t *testing.T) {
|
||||
if config.Log == nil {
|
||||
config.Log = zap.NewNop()
|
||||
t.Cleanup(func() { config.Log = nil })
|
||||
}
|
||||
// First poll sees RUN with the URL already missing (the state server
|
||||
// only exposes a URL once the state flips), second sees it present.
|
||||
// We emulate the missing-URL round by answering the first url query
|
||||
// with an empty URL: swap in a custom server.
|
||||
seen := 0
|
||||
emptyURL := true
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
url := ""
|
||||
if !emptyURL {
|
||||
url = "http://h:9000"
|
||||
}
|
||||
row := models.QueryJob{JobStatus: "RUN", InteractiveJobURL: url}
|
||||
if r.URL.Query().Get("fields") == "interactive_job_url" {
|
||||
seen++
|
||||
if seen == 1 {
|
||||
row.InteractiveJobURL = ""
|
||||
}
|
||||
emptyURL = row.InteractiveJobURL != ""
|
||||
}
|
||||
_ = json.NewEncoder(w).Encode(models.JobsResponse{
|
||||
Code: 0,
|
||||
Data: struct {
|
||||
Count int `json:"count"`
|
||||
Rows []models.QueryJob `json:"rows"`
|
||||
Page_token string `json:"page_token"`
|
||||
}{Rows: []models.QueryJob{row}},
|
||||
})
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
cli := NewJobSubmitter(nil)
|
||||
start := time.Now()
|
||||
url, err := cli.waitForInteractiveURL(request.NewAPIClient(srv.URL), "tok", "123")
|
||||
elapsed := time.Since(start)
|
||||
if err != nil {
|
||||
t.Fatalf("waitForInteractiveURL: %v", err)
|
||||
}
|
||||
if url != "http://h:9000" {
|
||||
t.Errorf("url = %q, want http://h:9000", url)
|
||||
}
|
||||
if elapsed > 1200*time.Millisecond {
|
||||
t.Errorf("elapsed = %v, want ~0.5s (run cadence, not the old 2s)", elapsed)
|
||||
}
|
||||
}
|
||||
|
||||
// A suspended interactive job must not fail the wait the way a terminal
|
||||
// state does: the allocation is held and `xce jobs resume` may follow, so
|
||||
// the loop keeps waiting (backed off) and picks up the URL when the job is
|
||||
// running again.
|
||||
func TestWaitForInteractiveURL_SuspendedJobKeepsWaiting(t *testing.T) {
|
||||
if config.Log == nil {
|
||||
config.Log = zap.NewNop()
|
||||
t.Cleanup(func() { config.Log = nil })
|
||||
}
|
||||
js := newJobStateServer(t, "USUSP", "RUN")
|
||||
cli := NewJobSubmitter(nil)
|
||||
url, err := cli.waitForInteractiveURL(request.NewAPIClient(js.srv.URL), "tok", "123")
|
||||
if err != nil {
|
||||
t.Fatalf("waitForInteractiveURL: %v", err)
|
||||
}
|
||||
if url != "http://h:9000" {
|
||||
t.Errorf("url = %q, want http://h:9000", url)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue