From 7b6f865b3526c7539f864f3d00e880e6026966e4 Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 26 Jul 2026 20:44:43 +0400 Subject: [PATCH] add admin diagnostics readiness and event metrics --- cmd/orchestra/main.go | 34 +++---- internal/admin/admin.go | 179 +++++++++++++++++++++++++++++++++++ internal/admin/admin_test.go | 52 ++++++++++ 3 files changed, 245 insertions(+), 20 deletions(-) create mode 100644 internal/admin/admin.go create mode 100644 internal/admin/admin_test.go diff --git a/cmd/orchestra/main.go b/cmd/orchestra/main.go index 931926c..d84a8ab 100644 --- a/cmd/orchestra/main.go +++ b/cmd/orchestra/main.go @@ -3,10 +3,10 @@ package main import ( "context" "encoding/json" - "fmt" "io" "log" "net/http" + "orchestra/internal/admin" "orchestra/internal/authz" "orchestra/internal/delivery" "orchestra/internal/domain" @@ -253,17 +253,18 @@ func main() { } json.NewEncoder(w).Encode(out) }) - mux.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { - tasks := s.Tasks() - counts := map[domain.TaskState]int{} - for _, t := range tasks { - counts[t.State]++ - } - w.Header().Set("Content-Type", "text/plain; version=0.0.4") - for _, state := range []domain.TaskState{domain.StateQueued, domain.StateLeased, domain.StateCompleted, domain.StateFailed, domain.StateBlocked} { - fmt.Fprintf(w, "orchestra_tasks{state=\"%s\"} %d\n", state, counts[state]) - } - }) + adminServer := &admin.Server{Store: s, RouterReady: rt != nil, Providers: providerHealth, Probes: map[string]admin.ProbeFunc{ + "router": func() (bool, string) { return rt != nil, "configured router" }, + "gitea": func() (bool, string) { + return os.Getenv("ORCHESTRA_GITEA_URL") == "" || providerHealth["gitea"] != nil, "configured provider" + }, + "jsonl": func() (bool, string) { + return os.Getenv("ORCHESTRA_JSONL") == "" || providerHealth["jsonl"] != nil, "configured provider" + }, + }} + mux.HandleFunc("/metrics", adminServer.Metrics) + mux.HandleFunc("/v1/events/subscribe", adminServer.Subscribe) + mux.HandleFunc("/v1/admin/diagnostics", adminServer.Diagnostics) mux.HandleFunc("/v1/tasks/", func(w http.ResponseWriter, r *http.Request) { parts := strings.Split(strings.Trim(r.URL.Path, "/"), "/") if len(parts) < 4 || len(parts) > 5 || r.Method != "POST" { @@ -407,14 +408,7 @@ func main() { }() } mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok\n")) }) - mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) { - checks := map[string]any{"store": true, "router": rt != nil, "gitea": os.Getenv("ORCHESTRA_GITEA_URL") != "", "jsonl": os.Getenv("ORCHESTRA_JSONL") != ""} - ready := rt != nil || (os.Getenv("ORCHESTRA_CONFIG") == "") - if !ready { - w.WriteHeader(http.StatusServiceUnavailable) - } - json.NewEncoder(w).Encode(map[string]any{"ready": ready, "checks": checks}) - }) + mux.HandleFunc("/readyz", adminServer.Readiness) mux.HandleFunc("/v1/providers/health", func(w http.ResponseWriter, r *http.Request) { out := map[string]provider.Health{} for name, sup := range providerHealth { diff --git a/internal/admin/admin.go b/internal/admin/admin.go new file mode 100644 index 0000000..e467945 --- /dev/null +++ b/internal/admin/admin.go @@ -0,0 +1,179 @@ +// 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/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 + 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) { + checks := []Probe{{Name: "store", Ready: s.Store != nil}} + if s.RouterReady { + checks = append(checks, Probe{Name: "router", Ready: true}) + } + 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{"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: + } + } +} diff --git a/internal/admin/admin_test.go b/internal/admin/admin_test.go new file mode 100644 index 0000000..f2e9062 --- /dev/null +++ b/internal/admin/admin_test.go @@ -0,0 +1,52 @@ +package admin + +import ( + "context" + "encoding/json" + "net/http/httptest" + "orchestra/internal/authz" + "orchestra/internal/domain" + "orchestra/internal/store" + "strings" + "testing" +) + +func TestDiagnosticsRequiresFullControl(t *testing.T) { + s, err := store.Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + h := (&Server{Store: s}).Diagnostics + r := httptest.NewRequest("GET", "/v1/admin/diagnostics", nil) + r.Header.Set("X-Orchestra-Surface", string(authz.MCP)) + w := httptest.NewRecorder() + h(w, r) + if w.Code != 403 || !strings.Contains(w.Body.String(), `"code":"forbidden"`) { + t.Fatalf("status=%d body=%s", w.Code, w.Body) + } +} + +func TestSubscribeEmitsCursorAndEvent(t *testing.T) { + s, err := store.Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + p, _ := json.Marshal(map[string]any{"source": "test", "external_id": "1", "project": "p"}) + if err := s.Append(domain.Event{ID: "e", TaskID: "t", Type: "TaskCreated", Version: 1, Payload: p}); err != nil { + t.Fatal(err) + } + h := (&Server{Store: s}).Subscribe + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + r := httptest.NewRequest("GET", "/v1/events/subscribe?since=0", nil).WithContext(ctx) + w := httptest.NewRecorder() + done := make(chan struct{}) + go func() { h(w, r); close(done) }() + for !strings.Contains(w.Body.String(), "id: 1") { + } + cancel() + <-done + if !strings.Contains(w.Body.String(), "id: 1") { + t.Fatalf("body=%s", w.Body) + } +}