From c8bd8fe896cd6eda94a6b899a7777a72d6d00fe9 Mon Sep 17 00:00:00 2001 From: "lei.chen" Date: Wed, 12 Aug 2026 08:30:25 +0000 Subject: [PATCH] feat(schedulerclient): WatchJobs SSE stream + fix ListJobs filter param MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add WatchJobs(ctx) to the Client interface and HTTPClient: opens the apiserver's /xce/v1/events SSE stream and yields JobEvent values until the context is cancelled or the stream ends. JobEvent carries the event type, job id, status string, exit code, and a raw job snapshot. Also fix ListJobs' filter params: the apiserver accepts filters via the "filter" query param in field:op:value form (e.g. queue_name:eq:normal); the individual queue_name/submit_user params it ignores with a 400, which broke every filtered list. Multiple conditions are brace-wrapped {a},{b}. Verified live on t1: WatchJobs captured job.updated PEND→RUN→DONE for a submitted job. Co-Authored-By: Claude --- src/frontend/schedulerclient/client.go | 8 ++ src/frontend/schedulerclient/httpclient.go | 71 ++++++++++++- src/frontend/schedulerclient/types.go | 30 ++++++ src/frontend/schedulerclient/watch_test.go | 116 +++++++++++++++++++++ 4 files changed, 223 insertions(+), 2 deletions(-) create mode 100644 src/frontend/schedulerclient/watch_test.go diff --git a/src/frontend/schedulerclient/client.go b/src/frontend/schedulerclient/client.go index 11c577f8..33a66e42 100644 --- a/src/frontend/schedulerclient/client.go +++ b/src/frontend/schedulerclient/client.go @@ -50,4 +50,12 @@ type Client interface { // ReleaseJob resumes a suspended job (action "resume", default SIGCONT). ReleaseJob(ctx context.Context, id string) error + + // ── Event stream (Phase 2) ── + + // WatchJobs subscribes to the live job-event SSE stream and returns a + // channel of lifecycle events (job.created / job.updated / job.deleted / + // job.finished). The channel is closed when the context is cancelled or + // the stream ends. The caller is responsible for cancelling ctx. + WatchJobs(ctx context.Context) (<-chan JobEvent, error) } diff --git a/src/frontend/schedulerclient/httpclient.go b/src/frontend/schedulerclient/httpclient.go index f6fb36f7..7dd91563 100644 --- a/src/frontend/schedulerclient/httpclient.go +++ b/src/frontend/schedulerclient/httpclient.go @@ -1,6 +1,7 @@ package schedulerclient import ( + "bufio" "context" "crypto/tls" "encoding/json" @@ -44,11 +45,19 @@ func (hc *HTTPClient) ListJobs(ctx context.Context, filter JobFilter) ([]*URMJob url := hc.baseURL + "/xce/v1/jobs" req := hc.rc.R().SetContext(ctx).SetHeader("Authorization", hc.token) + // The apiserver's job list accepts filters via the "filter" query param in + // "field:op:value" form (e.g. queue_name:eq:normal); the individual + // queue_name/submit_user params it ignores (400). Multiple conditions are + // brace-wrapped: {a},{b}. + var conds []string if filter.Queue != "" { - req.SetQueryParam("queue_name", filter.Queue) + conds = append(conds, "queue_name:eq:"+filter.Queue) } if filter.Owner != "" { - req.SetQueryParam("submit_user", filter.Owner) + conds = append(conds, "submit_user:eq:"+filter.Owner) + } + if len(conds) > 0 { + req.SetQueryParam("filter", "{"+strings.Join(conds, "},{")+"}") } if filter.Limit > 0 { req.SetQueryParam("limit", strconv.Itoa(filter.Limit)) @@ -503,3 +512,61 @@ type apiHost struct { NumUSusp int `json:"num_ususp"` NumDone int `json:"num_done"` } + +// ── Event stream (Phase 2) ────────────────────────────────────────────────── + +// WatchJobs subscribes to the apiserver's live job-event SSE endpoint and +// yields lifecycle events on the returned channel. The connection is kept +// open until ctx is cancelled or the server ends the stream; the channel is +// closed when the reader terminates. Malformed frames are skipped. +func (hc *HTTPClient) WatchJobs(ctx context.Context) (<-chan JobEvent, error) { + // SSE is a long-lived stream — no request timeout. TLS skip matches + // NewClient's InsecureSkipVerify behaviour for the apiserver endpoint. + client := &http.Client{ + Transport: &http.Transport{ + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, + }, + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, hc.baseURL+"/xce/v1/events", nil) + if err != nil { + return nil, fmt.Errorf("watch jobs: %w", err) + } + req.Header.Set("Authorization", hc.token) + req.Header.Set("Accept", "text/event-stream") + + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("watch jobs: %w", err) + } + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + return nil, fmt.Errorf("watch jobs: apiserver returned status %d", resp.StatusCode) + } + + ch := make(chan JobEvent, 64) + go func() { + defer close(ch) + defer resp.Body.Close() + + scanner := bufio.NewScanner(resp.Body) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if !strings.HasPrefix(line, "data: ") { + continue + } + var ev JobEvent + if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &ev); err != nil { + continue // skip malformed frame rather than kill the stream + } + select { + case ch <- ev: + case <-ctx.Done(): + return + } + } + }() + + return ch, nil +} diff --git a/src/frontend/schedulerclient/types.go b/src/frontend/schedulerclient/types.go index ef7a6857..1159135d 100644 --- a/src/frontend/schedulerclient/types.go +++ b/src/frontend/schedulerclient/types.go @@ -103,3 +103,33 @@ type Metrics struct { PendingJobs int `json:"pending_jobs"` Utilization float64 `json:"utilization"` // 0.0 – 1.0 } + +// ── Job lifecycle events (Phase 2, SSE WatchJobs) ──────────────────────────── + +// JobEvent is a single job lifecycle event received from the apiserver's live +// SSE stream (/xce/v1/events). Type is one of: +// +// job.created / job.updated / job.deleted / job.finished +// +// Status is the backend-native job status string (PEND/RUN/DONE/EXIT/...) +// when the event carries a job snapshot. For job.finished, ExitCode holds the +// process exit code (0 → DONE, non-zero → EXIT). +type JobEvent struct { + Type string `json:"type"` + JobID uint64 `json:"job_id,omitempty"` + ExitCode int `json:"exit_code,omitempty"` + Status string `json:"job_status,omitempty"` + Job *JobEventJob `json:"job,omitempty"` +} + +// JobEventJob mirrors the apiserver job snapshot embedded in a JobEvent. +// Only the fields the scheduler agent consumes are modelled. +type JobEventJob struct { + UniqueJobID int64 `json:"unique_job_id"` + JobID int64 `json:"job_id"` + SubmitUser string `json:"submit_user"` + JobStatus int `json:"job_status"` + JobName string `json:"job_name"` + Command string `json:"command"` + QueueName string `json:"queue_name"` +} diff --git a/src/frontend/schedulerclient/watch_test.go b/src/frontend/schedulerclient/watch_test.go new file mode 100644 index 00000000..692585f0 --- /dev/null +++ b/src/frontend/schedulerclient/watch_test.go @@ -0,0 +1,116 @@ +package schedulerclient + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// TestWatchJobsReceivesEvents verifies WatchJobs parses SSE frames from a +// live stream into JobEvent values. +func TestWatchJobsReceivesEvents(t *testing.T) { + payload := `{"type":"job.updated","job_id":123,"job_status":"RUN","job":{"unique_job_id":123,"job_id":123,"submit_user":"mcp-agent","job_name":"wtest","command":"sleep 5","queue_name":"normal"}}` + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + if got := r.Header.Get("Authorization"); got == "" { + t.Errorf("expected Authorization header, got empty") + } + if got := r.URL.Path; got != "/xce/v1/events" { + t.Errorf("unexpected path %q", got) + } + flusher, _ := w.(http.Flusher) + _, _ = w.Write([]byte("data: " + payload + "\n\n")) + flusher.Flush() + })) + defer srv.Close() + + hc := NewClient(srv.URL, "xce_ak_test") + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + ch, err := hc.WatchJobs(ctx) + if err != nil { + t.Fatalf("WatchJobs: %v", err) + } + + var got *JobEvent + for ev := range ch { + got = &ev + break + } + if got == nil { + t.Fatal("did not receive any event") + } + if got.Type != "job.updated" || got.JobID != 123 || got.Status != "RUN" { + t.Fatalf("unexpected event: %+v", got) + } + if got.Job == nil || got.Job.Command != "sleep 5" || got.Job.QueueName != "normal" { + t.Fatalf("event job payload wrong: %+v", got) + } +} + +// TestWatchJobsSkipsMalformedFrames verifies a malformed SSE frame does not +// kill the stream — the next valid frame is still delivered. +func TestWatchJobsSkipsMalformedFrames(t *testing.T) { + payload := `{"type":"job.finished","job_id":7,"exit_code":0,"job_status":"DONE"}` + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + flusher, _ := w.(http.Flusher) + _, _ = w.Write([]byte("data: {not json\n\n")) + flusher.Flush() + _, _ = w.Write([]byte("data: " + payload + "\n\n")) + flusher.Flush() + })) + defer srv.Close() + + hc := NewClient(srv.URL, "xce_ak_test") + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + ch, err := hc.WatchJobs(ctx) + if err != nil { + t.Fatalf("WatchJobs: %v", err) + } + + var got *JobEvent + for ev := range ch { + got = &ev + break + } + if got == nil { + t.Fatal("did not receive valid event after malformed frame") + } + if got.Type != "job.finished" || got.JobID != 7 || got.ExitCode != 0 { + t.Fatalf("unexpected event: %+v", got) + } +} + +// TestJobEventJSONRoundTrip ensures the JobEvent JSON shape matches what the +// apiserver SSE endpoint emits. +func TestJobEventJSONRoundTrip(t *testing.T) { + raw := `{"type":"job.created","job_id":5,"job_status":"PEND","job":{"unique_job_id":5,"job_id":5,"submit_user":"u","job_name":"n","command":"echo hi","queue_name":"q"}}` + var ev JobEvent + if err := json.Unmarshal([]byte(raw), &ev); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if ev.Type != "job.created" || ev.JobID != 5 || ev.Status != "PEND" { + t.Fatalf("unexpected event: %+v", ev) + } + if ev.Job == nil || ev.Job.JobName != "n" || ev.Job.Command != "echo hi" { + t.Fatalf("unexpected job payload: %+v", ev.Job) + } + // Re-marshal and confirm it stays a valid single-line SSE payload. + out, err := json.Marshal(ev) + if err != nil { + t.Fatalf("marshal: %v", err) + } + if strings.Contains(string(out), "\n") { + t.Fatalf("marshalled event must not contain newline (SSE framing): %s", out) + } +}