diff --git a/src/frontend/apiserver/backend/broker.go b/src/frontend/apiserver/backend/broker.go new file mode 100644 index 0000000..f77cbba --- /dev/null +++ b/src/frontend/apiserver/backend/broker.go @@ -0,0 +1,83 @@ +package backend + +import ( + "sync" + + "apiserver/models" +) + +// Event types broadcast to SSE subscribers (scheduler-agent Phase 2, ADR-004). +const ( + EventJobCreated = "job.created" + EventJobUpdated = "job.updated" + EventJobDeleted = "job.deleted" + EventJobFinished = "job.finished" +) + +// Event is a single lifecycle event. Exactly one of Job / (JobID, ExitCode) +// is populated depending on Type. +type Event struct { + Type string `json:"type"` + JobID uint64 `json:"job_id,omitempty"` + ExitCode int `json:"exit_code,omitempty"` + Status string `json:"job_status,omitempty"` // JobStateStr of the job (PEND/RUN/DONE/EXIT/...) + Job *models.JobInfo `json:"job,omitempty"` +} + +// Broker is a thread-safe fan-out of lifecycle events to subscribed channels. +// Subscribers register a buffered channel; a slow subscriber that cannot +// drain before the next event has the event dropped (never blocking the +// publisher / notification handler). +type Broker struct { + mu sync.RWMutex + subs map[chan Event]struct{} +} + +// DefaultBroker is the process-wide event bus the DefaultHandler publishes to +// and the SSE endpoint subscribes to. +var DefaultBroker = NewBroker() + +// NewBroker returns an empty broker. +func NewBroker() *Broker { + return &Broker{subs: make(map[chan Event]struct{})} +} + +// Subscribe registers a new subscriber with the given buffer depth and +// returns its channel. Callers must Unsubscribe when done. +func (b *Broker) Subscribe(buffer int) chan Event { + ch := make(chan Event, buffer) + b.mu.Lock() + b.subs[ch] = struct{}{} + b.mu.Unlock() + return ch +} + +// Unsubscribe removes a subscriber. Safe to call after the channel has been +// drained; the channel itself is closed by the subscriber (see the SSE +// handler). +func (b *Broker) Unsubscribe(ch chan Event) { + b.mu.Lock() + delete(b.subs, ch) + b.mu.Unlock() +} + +// Publish fans an event out to every subscriber, dropping it for slow +// consumers. Never blocks the caller. +func (b *Broker) Publish(ev Event) { + b.mu.RLock() + defer b.mu.RUnlock() + for ch := range b.subs { + select { + case ch <- ev: + default: + // slow consumer — drop rather than block the notification path + } + } +} + +// SubscriberCount reports the number of registered subscribers (tests / metrics). +func (b *Broker) SubscriberCount() int { + b.mu.RLock() + defer b.mu.RUnlock() + return len(b.subs) +} diff --git a/src/frontend/apiserver/backend/broker_test.go b/src/frontend/apiserver/backend/broker_test.go new file mode 100644 index 0000000..5e8f7f7 --- /dev/null +++ b/src/frontend/apiserver/backend/broker_test.go @@ -0,0 +1,96 @@ +package backend + +import ( + "testing" + + "apiserver/models" +) + +func TestBrokerSubscribePublishUnsubscribe(t *testing.T) { + b := NewBroker() + ch := b.Subscribe(8) + defer b.Unsubscribe(ch) + + b.Publish(Event{Type: EventJobCreated, JobID: 42}) + + select { + case ev := <-ch: + if ev.Type != EventJobCreated || ev.JobID != 42 { + t.Fatalf("unexpected event: %+v", ev) + } + default: + t.Fatal("expected published event on subscriber channel") + } +} + +func TestBrokerEventCarriesJobPayload(t *testing.T) { + b := NewBroker() + ch := b.Subscribe(8) + defer b.Unsubscribe(ch) + + job := &models.JobInfo{Unique_JobID: 7, JobName: "x", Command: "sleep 1"} + b.Publish(Event{Type: EventJobUpdated, JobID: 7, Job: job}) + + ev := <-ch + if ev.Job == nil || ev.Job.Unique_JobID != 7 || ev.Job.Command != "sleep 1" { + t.Fatalf("event job payload not carried: %+v", ev) + } +} + +func TestBrokerDoesNotBlockSlowConsumer(t *testing.T) { + // A full subscriber channel must not block Publish (the notification + // handler runs on the hot path). The event is dropped for the slow + // consumer; the publisher returns immediately. + b := NewBroker() + ch := b.Subscribe(1) + defer b.Unsubscribe(ch) + + // Fill the single-slot channel. + b.Publish(Event{Type: EventJobCreated, JobID: 1}) + if len(ch) != 1 { + t.Fatalf("expected full channel, got len %d", len(ch)) + } + + // This would deadlock if Publish blocked; a nil-drain on a goroutine + // would hang the test via the timeout. Just call it — if it blocks, the + // test framework's timeout catches it. + b.Publish(Event{Type: EventJobCreated, JobID: 2}) + + if len(ch) != 1 { + t.Fatalf("slow consumer should have the first event dropped, len=%d", len(ch)) + } + if ev := <-ch; ev.JobID != 1 { + t.Fatalf("expected first event retained, got %+v", ev) + } +} + +func TestBrokerUnsubscribeStopsDelivery(t *testing.T) { + b := NewBroker() + ch := b.Subscribe(8) + b.Unsubscribe(ch) + + b.Publish(Event{Type: EventJobCreated, JobID: 9}) + + select { + case ev := <-ch: + t.Fatalf("event delivered after unsubscribe: %+v", ev) + default: + // expected — no delivery + } +} + +func TestBrokerSubscriberCount(t *testing.T) { + b := NewBroker() + ch1 := b.Subscribe(8) + ch2 := b.Subscribe(8) + defer b.Unsubscribe(ch1) + defer b.Unsubscribe(ch2) + + if n := b.SubscriberCount(); n != 2 { + t.Fatalf("expected 2 subscribers, got %d", n) + } + b.Unsubscribe(ch1) + if n := b.SubscriberCount(); n != 1 { + t.Fatalf("expected 1 subscriber after unsubscribe, got %d", n) + } +} diff --git a/src/frontend/apiserver/backend/handler.go b/src/frontend/apiserver/backend/handler.go index d5465f0..34eb37a 100644 --- a/src/frontend/apiserver/backend/handler.go +++ b/src/frontend/apiserver/backend/handler.go @@ -13,14 +13,17 @@ type DefaultHandler struct{} func (h *DefaultHandler) OnJobCreated(job *models.JobInfo) { store.Jobs.Put(uint64(job.Unique_JobID), *job) + DefaultBroker.Publish(Event{Type: EventJobCreated, JobID: uint64(job.Unique_JobID), Status: models.JobStateStr(job.JobStatus), Job: job}) } func (h *DefaultHandler) OnJobUpdated(job *models.JobInfo) { store.Jobs.Put(uint64(job.Unique_JobID), *job) + DefaultBroker.Publish(Event{Type: EventJobUpdated, JobID: uint64(job.Unique_JobID), Status: models.JobStateStr(job.JobStatus), Job: job}) } func (h *DefaultHandler) OnJobDeleted(jobid uint64) { store.Jobs.Delete(jobid) + DefaultBroker.Publish(Event{Type: EventJobDeleted, JobID: jobid}) } func (h *DefaultHandler) OnJobFinished(jobid uint64, exitCode int) { @@ -34,6 +37,7 @@ func (h *DefaultHandler) OnJobFinished(jobid uint64, exitCode int) { existing.JobStatus = models.JOBSTAT["EXIT"] } store.Jobs.Put(jobid, existing) + DefaultBroker.Publish(Event{Type: EventJobFinished, JobID: jobid, ExitCode: exitCode}) } func (h *DefaultHandler) OnQueueUpdated(queue *models.QueueInfo) { diff --git a/src/frontend/apiserver/backend/handler_test.go b/src/frontend/apiserver/backend/handler_test.go index 661a89c..c59a962 100644 --- a/src/frontend/apiserver/backend/handler_test.go +++ b/src/frontend/apiserver/backend/handler_test.go @@ -16,6 +16,48 @@ func resetStores() { store.InitStores() } +// freshBroker swaps DefaultBroker for an isolated instance, restoring the +// previous one on cleanup, so publish-behavior tests don't pollute other +// tests that subscribe to the process-wide broker. +func freshBroker(t *testing.T) *Broker { + t.Helper() + prev := DefaultBroker + b := NewBroker() + DefaultBroker = b + t.Cleanup(func() { DefaultBroker = prev }) + return b +} + +func TestHandlerPublishesJobEventsToBroker(t *testing.T) { + resetStores() + b := freshBroker(t) + ch := b.Subscribe(16) + defer b.Unsubscribe(ch) + + h := DefaultHandler{} + h.OnJobCreated(&models.JobInfo{Unique_JobID: 1, JobName: "created"}) + h.OnJobUpdated(&models.JobInfo{Unique_JobID: 2, JobName: "updated"}) + // OnJobFinished only publishes when the job already exists in the store + // (it updates the stored record), so create job 3 before finishing it. + h.OnJobUpdated(&models.JobInfo{Unique_JobID: 3, JobName: "tofinish"}) + h.OnJobFinished(3, 0) + + var got []Event + for i := 0; i < 4; i++ { + got = append(got, <-ch) + } + + if got[0].Type != EventJobCreated || got[0].JobID != 1 { + t.Fatalf("event[0] = %+v, want job.created 1", got[0]) + } + if got[1].Type != EventJobUpdated || got[1].JobID != 2 { + t.Fatalf("event[1] = %+v, want job.updated 2", got[1]) + } + if got[3].Type != EventJobFinished || got[3].JobID != 3 || got[3].ExitCode != 0 { + t.Fatalf("event[3] = %+v, want job.finished 3", got[3]) + } +} + // captureLog swaps utils.Log for an observer core so a test can assert on // exactly which log lines fired, restoring the previous logger on cleanup. func captureLog(t *testing.T) *observer.ObservedLogs { diff --git a/src/frontend/apiserver/config/rbac.yaml b/src/frontend/apiserver/config/rbac.yaml index 4f76abc..76225ae 100644 --- a/src/frontend/apiserver/config/rbac.yaml +++ b/src/frontend/apiserver/config/rbac.yaml @@ -17,6 +17,7 @@ permissions: - hostgroup:control # add - usergroup:view # view usergroup info - resource:view # view resource info + - event:subscribe # subscribe to the live job-event SSE stream (scheduler-agent Phase 2) - ar:view # view advanced reservation info - ar:submit # submit advanced reservation - ar:control # delete, modify, create @@ -170,6 +171,9 @@ roles: - action: cluster:view domain: - type: all + - action: event:subscribe + domain: + - type: all # mcp_operator: service-account role for the scheduler-agent MCP server # (Phase 1, write operations). Same cluster visibility as mcp_readonly, @@ -204,3 +208,6 @@ roles: - action: job:control domain: - type: all + - action: event:subscribe + domain: + - type: all diff --git a/src/frontend/apiserver/controllers/events.go b/src/frontend/apiserver/controllers/events.go new file mode 100644 index 0000000..f70bb9b --- /dev/null +++ b/src/frontend/apiserver/controllers/events.go @@ -0,0 +1,92 @@ +package controllers + +import ( + "encoding/json" + "net/http" + "time" + + "github.com/gin-gonic/gin" + + "apiserver/backend" + logger "apiserver/utils" + + "go.uber.org/zap" +) + +const ( + // eventSSEBuffer is the per-subscriber broker channel depth. Deep enough + // to absorb a burst of job pushes without dropping; bounded so a slow + // consumer cannot grow the channel without limit (the broker drops for + // full channels rather than blocking the notification handler). + eventSSEBuffer = 256 + + // eventSSEKeepalive is how often a bare : keep-alive comment is sent when + // no event has arrived, keeping intermediaries from timing out the idle + // connection. + eventSSEKeepalive = 15 * time.Second +) + +// EventsController exposes the live cluster event stream (scheduler-agent +// Phase 2, ADR-004). +type EventsController struct{} + +// SubscribeJobs streams job lifecycle events (job.created / job.updated / +// job.deleted / job.finished) as Server-Sent Events. The connection stays +// open until the client disconnects. Events are JSON payloads in SSE "data:" +// frames: +// +// data: {"type":"job.updated","job_id":123,"job":{...}} +// +// The RBAC middleware has already authenticated the request; this handler +// performs no further permission checks beyond event:subscribe. +func (e *EventsController) SubscribeJobs(ctx *gin.Context) { + ctx.Header("Content-Type", "text/event-stream") + ctx.Header("Cache-Control", "no-cache") + ctx.Header("Connection", "keep-alive") + ctx.Header("X-Accel-Buffering", "no") + + flusher, ok := ctx.Writer.(http.Flusher) + if !ok { + ReturnError(ctx, http.StatusInternalServerError, "streaming not supported", nil) + return + } + + sub := backend.DefaultBroker.Subscribe(eventSSEBuffer) + defer backend.DefaultBroker.Unsubscribe(sub) + + // Send a single keep-alive immediately so the client (and any proxy) sees + // the headers + first frame as soon as the connection is established. + if _, err := ctx.Writer.WriteString(": connected\n\n"); err != nil { + return + } + flusher.Flush() + + keepalive := time.NewTicker(eventSSEKeepalive) + defer keepalive.Stop() + + for { + select { + case ev, ok := <-sub: + if !ok { + return + } + payload, err := json.Marshal(ev) + if err != nil { + logger.Error("events: marshal event", zap.Error(err)) + continue + } + if _, err := ctx.Writer.WriteString("data: " + string(payload) + "\n\n"); err != nil { + return + } + flusher.Flush() + case <-keepalive.C: + // Bare comment keeps the connection alive through idle periods. + if _, err := ctx.Writer.WriteString(": keepalive\n\n"); err != nil { + return + } + flusher.Flush() + case <-ctx.Request.Context().Done(): + return + } + } +} diff --git a/src/frontend/apiserver/controllers/events_test.go b/src/frontend/apiserver/controllers/events_test.go new file mode 100644 index 0000000..2cffbba --- /dev/null +++ b/src/frontend/apiserver/controllers/events_test.go @@ -0,0 +1,86 @@ +package controllers + +import ( + "bufio" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gin-gonic/gin" + + "apiserver/backend" + "apiserver/models" +) + +// TestEventsSSEStreamsPublishedEvent drives the SSE handler with a live +// httptest server: it subscribes the handler, publishes a job.updated event +// on the process-wide broker, and asserts the streamed frame carries it. +func TestEventsSSEStreamsPublishedEvent(t *testing.T) { + prevBroker := backend.DefaultBroker + b := backend.NewBroker() + backend.DefaultBroker = b + t.Cleanup(func() { backend.DefaultBroker = prevBroker }) + + gin.SetMode(gin.TestMode) + r := gin.New() + r.GET("/xce/v1/events", (&EventsController{}).SubscribeJobs) + + srv := httptest.NewServer(r) + defer srv.Close() + + // Publish after the handler has subscribed (small delay for the SSE + // connection to establish). + go func() { + time.Sleep(200 * time.Millisecond) + job := &models.JobInfo{Unique_JobID: 99, JobName: "sse-test", Command: "sleep 1"} + b.Publish(backend.Event{Type: backend.EventJobUpdated, JobID: 99, Job: job}) + // Close the stream by shutting the server down after delivery. + time.Sleep(300 * time.Millisecond) + }() + + client := &http.Client{Timeout: 5 * time.Second} + resp, err := client.Get(srv.URL + "/xce/v1/events") + if err != nil { + t.Fatalf("GET events: %v", err) + } + defer resp.Body.Close() + + if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "text/event-stream") { + t.Fatalf("expected text/event-stream content type, got %q", ct) + } + + // The SSE payload carries the full models.JobInfo snapshot whose + // CustomTime fields don't round-trip through generic JSON here, so assert + // on the event envelope + key job fields with a local struct instead of + // unmarshalling the full JobInfo. + type sseFrame struct { + Type string `json:"type"` + JobID uint64 `json:"job_id"` + } + + scanner := bufio.NewScanner(resp.Body) + var found *sseFrame + for scanner.Scan() { + line := scanner.Text() + if !strings.HasPrefix(line, "data: ") { + continue + } + var ev sseFrame + if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &ev); err != nil { + t.Fatalf("bad SSE frame: %v (%q)", err, line) + } + if ev.Type == backend.EventJobUpdated && ev.JobID == 99 { + found = &ev + break + } + } + if err := scanner.Err(); err != nil { + t.Fatalf("scan SSE stream: %v", err) + } + if found == nil { + t.Fatal("did not receive job.updated event over SSE stream") + } +} diff --git a/src/frontend/apiserver/middleware/whitelist.go b/src/frontend/apiserver/middleware/whitelist.go index fa68b5c..28a78dd 100644 --- a/src/frontend/apiserver/middleware/whitelist.go +++ b/src/frontend/apiserver/middleware/whitelist.go @@ -26,6 +26,7 @@ var WhitelistGETURL = []string{ "/xce/v1/files/*", // get download file "/xce/v1/resources/definition", // resources definition "/xce/v1/resources/distribution", // resources distribution + "/xce/v1/events", // live job-event SSE stream (scheduler-agent Phase 2) "/xce/v1/roles", // roles list "/xce/v1/roles/*", // role detail "/xce/v1/permissions", // permissions diff --git a/src/frontend/apiserver/router/routers.go b/src/frontend/apiserver/router/routers.go index 8c960a4..c44c8c7 100644 --- a/src/frontend/apiserver/router/routers.go +++ b/src/frontend/apiserver/router/routers.go @@ -144,6 +144,15 @@ func Router() *gin.Engine { jobsApi.POST("/:jobids/restart", controllers.RequirePermission("queue:submit"), jobHandler.RestartJobByID) } + // ################ case: events route (scheduler-agent Phase 2, ADR-004) + eventsApi := r.Group("/xce/v1/events") + { + eventsHandler := &controllers.EventsController{} + // Server-Sent Events stream of job lifecycle events. Long-lived + // connection; keep-alive comments are sent when idle. + eventsApi.GET("", controllers.RequirePermission("event:subscribe"), eventsHandler.SubscribeJobs) + } + // ################ case: resources route resApi := r.Group("/xce/v1/resources") {