add admin diagnostics readiness and event metrics

2026-07-26 20:44:43 +04:00
parent c3d8271e15
commit 7b6f865b35
3 changed files with 245 additions and 20 deletions
@@ -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 {
@@ -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:
+		}
+	}
+}
@@ -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)
+	}
+}