// Package admin contains the operational HTTP surface. It deliberately keeps // diagnostics and probes separate from task mutation handlers. package admin import ( "encoding/json" "errors" "fmt" "net/http" "orchestra/internal/authz" "orchestra/internal/buildinfo" "orchestra/internal/domain" "orchestra/internal/provider" "orchestra/internal/store" "strconv" "time" ) const MaxJSONBody = 256 << 10 type Error struct { Error string `json:"error"` Code string `json:"code"` } func WriteError(w http.ResponseWriter, status int, code string, err error) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) _ = json.NewEncoder(w).Encode(Error{Error: err.Error(), Code: code}) } func Decode(w http.ResponseWriter, r *http.Request, dst any) error { r.Body = http.MaxBytesReader(w, r.Body, MaxJSONBody) dec := json.NewDecoder(r.Body) dec.DisallowUnknownFields() if err := dec.Decode(dst); err != nil { return fmt.Errorf("invalid request: %w", err) } var extra any if err := dec.Decode(&extra); err == nil { return errors.New("invalid request: multiple JSON values") } return nil } type Probe struct { Name string `json:"name"` Ready bool `json:"ready"` Detail string `json:"detail,omitempty"` } type ProbeFunc func() (bool, string) type Server struct { Store *store.Store RouterReady bool Build buildinfo.Info Probes map[string]ProbeFunc Providers map[string]*provider.Supervisor } func (s *Server) authorize(r *http.Request) error { surface := authz.ParseSurface(r.Header.Get("X-Orchestra-Surface")) if surface == "" { surface = authz.Web } if authz.CapabilityFor(surface) != authz.FullControl { return errors.New("admin control requires full-control surface") } return nil } func (s *Server) Readiness(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") checks := []Probe{{Name: "store", Ready: s.Store != nil}} checks = append(checks, Probe{Name: "router", Ready: s.RouterReady}) for name, probe := range s.Probes { ok, detail := probe() checks = append(checks, Probe{Name: name, Ready: ok, Detail: detail}) } ready := len(checks) > 0 for _, c := range checks { ready = ready && c.Ready } if !ready { w.WriteHeader(http.StatusServiceUnavailable) } _ = json.NewEncoder(w).Encode(map[string]any{"ready": ready, "checks": checks}) } func (s *Server) Diagnostics(w http.ResponseWriter, r *http.Request) { if err := s.authorize(r); err != nil { WriteError(w, http.StatusForbidden, "forbidden", err) return } if s.Store == nil { WriteError(w, 500, "store_unavailable", errors.New("store unavailable")) return } tasks := s.Store.Tasks() events := s.Store.Events(0) _ = json.NewEncoder(w).Encode(map[string]any{"build": s.Build, "tasks": len(tasks), "events": len(events), "last_seq": func() uint64 { if len(events) == 0 { return 0 } return events[len(events)-1].Seq }(), "providers": s.providers()}) } func (s *Server) providers() map[string]provider.Health { out := map[string]provider.Health{} for n, p := range s.Providers { out[n] = p.Health() } return out } func (s *Server) Metrics(w http.ResponseWriter, r *http.Request) { if s.Store == nil { WriteError(w, 500, "store_unavailable", errors.New("store unavailable")) return } counts := map[domain.TaskState]int{} for _, t := range s.Store.Tasks() { counts[t.State]++ } events := s.Store.Events(0) types := map[string]int{} for _, e := range events { types[e.Type]++ } w.Header().Set("Content-Type", "text/plain; version=0.0.4") for _, st := range []domain.TaskState{domain.StateQueued, domain.StateLeased, domain.StateCompleted, domain.StateFailed, domain.StateBlocked} { fmt.Fprintf(w, "orchestra_tasks{state=\"%s\"} %d\n", st, counts[st]) } fmt.Fprintf(w, "orchestra_events_total %d\n", len(events)) for typ, n := range types { fmt.Fprintf(w, "orchestra_events{type=\"%s\"} %d\n", typ, n) } for n, h := range s.providers() { v := 0 if h.Running { v = 1 } fmt.Fprintf(w, "orchestra_provider_running{provider=\"%s\"} %d\n", n, v) } } func (s *Server) Subscribe(w http.ResponseWriter, r *http.Request) { if s.Store == nil { WriteError(w, 500, "store_unavailable", errors.New("store unavailable")) return } cursor, _ := strconv.ParseUint(r.URL.Query().Get("since"), 10, 64) timeout := time.NewTimer(15 * time.Second) defer timeout.Stop() ticker := time.NewTicker(100 * time.Millisecond) defer ticker.Stop() w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") f, ok := w.(http.Flusher) if !ok { WriteError(w, 500, "stream_unsupported", errors.New("stream unsupported")) return } for { es := s.Store.Events(cursor) for _, e := range es { b, _ := json.Marshal(e) fmt.Fprintf(w, "id: %d\ndata: %s\n\n", e.Seq, b) cursor = e.Seq f.Flush() } select { case <-r.Context().Done(): return case <-timeout.C: return case <-ticker.C: } } }