a81bbb2bcf
---ci--- project: orca phase: 9 milestone: v0.12 status: execute ---/ci--- - Start() refuses plaintext mode (mTLS required, R-021/REQ-123). - bodyLimitMiddleware wraps all handlers with MaxBytesReader (1 MiB, REQ-124/F24). - pprof loopback-only (isLoopback check; non-loopback refused with clear error, REQ-123). 2 new pprof loopback tests + existing daemon tests pass. Full build + vet green.
200 lines
6.5 KiB
Go
200 lines
6.5 KiB
Go
// Package daemon implements the orca HTTP daemon.
|
|
//
|
|
// The daemon exposes health endpoints (/healthz, /readyz), a status endpoint
|
|
// (/v1/status), and a v1 resource API for jobs, nodes, and tasks. All handlers
|
|
// follow the project conventions:
|
|
//
|
|
// - context.Context propagated to all I/O
|
|
// - errors wrapped with %w
|
|
// - structured JSON via writeJSON
|
|
// - no secrets in logs
|
|
// - input validation on path/query/body
|
|
//
|
|
// Deprecated: v0.9 re-architecture replaces this with SSH-push to bare
|
|
// servers (no orca binary on servers) per R-001. The orca daemon is
|
|
// repurposed to drain-and-stop in v0.10-P05 and scheduled for deletion
|
|
// in v0.10-P14. See .ciagent/PRD_v0.9.md R-001/R-006. The dual-write
|
|
// window (REQ-090/REQ-085) keeps this package compiling until v0.10-P14.
|
|
package daemon
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"net/http"
|
|
"sync/atomic"
|
|
"time"
|
|
)
|
|
|
|
// Server is the orca HTTP daemon. It holds shared dependencies and lifecycle
|
|
// state. Construct it with NewServer, then call Start/Shutdown.
|
|
type Server struct {
|
|
db *sql.DB
|
|
log *slog.Logger
|
|
addr string
|
|
ready atomic.Bool
|
|
|
|
httpServer *http.Server
|
|
pprofServer *http.Server
|
|
|
|
// mtls is non-nil after StartMTLS has been called; nil otherwise.
|
|
// Plaintext HTTP and mTLS are mutually exclusive — a Server is
|
|
// either in plaintext mode (default, v0.1 compat) or mTLS mode
|
|
// (v0.2 P01 forward).
|
|
mtls *MTLSState
|
|
|
|
// dispatch is the orca.v1.Dispatch service mounted on
|
|
// /orca.v1.Dispatch/* (P02). Optional — nil if no Dispatcher
|
|
// was registered. P02 wires this via RegisterDispatch.
|
|
dispatch *DispatchHandlers
|
|
}
|
|
|
|
// Options configures a new Server.
|
|
type Options struct {
|
|
DB *sql.DB
|
|
Log *slog.Logger
|
|
Addr string
|
|
Actor string // used for audit logging from API requests
|
|
|
|
// PprofAddr enables the pprof endpoint on a separate listener
|
|
// when non-empty (e.g. "127.0.0.1:6060"). Default "" disables it.
|
|
// The pprof listener is unauthenticated and operator-only; never
|
|
// expose it publicly (AD-024).
|
|
PprofAddr string
|
|
}
|
|
|
|
// maxBodyBytes is the limit for request bodies on JSON-decoding
|
|
// endpoints (REQ-124, F24). 1 MiB is generous for orca API calls.
|
|
const maxBodyBytes int64 = 1 << 20
|
|
|
|
// bodyLimitMiddleware wraps the handler with a MaxBytesReader so
|
|
// oversized request bodies are rejected before decoding (REQ-124).
|
|
func bodyLimitMiddleware(next http.Handler) http.Handler {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes)
|
|
next.ServeHTTP(w, r)
|
|
})
|
|
}
|
|
|
|
// NewServer constructs a Server with the default mux and route table.
|
|
func NewServer(opts Options) *Server {
|
|
if opts.Log == nil {
|
|
opts.Log = slog.Default()
|
|
}
|
|
if opts.Addr == "" {
|
|
opts.Addr = ":8080"
|
|
}
|
|
if opts.Actor == "" {
|
|
opts.Actor = "api"
|
|
}
|
|
s := &Server{
|
|
db: opts.DB,
|
|
log: opts.Log,
|
|
addr: opts.Addr,
|
|
}
|
|
s.httpServer = &http.Server{
|
|
Addr: opts.Addr,
|
|
Handler: s.mux(),
|
|
ReadHeaderTimeout: 5 * time.Second,
|
|
ReadTimeout: 15 * time.Second,
|
|
WriteTimeout: 30 * time.Second,
|
|
IdleTimeout: 60 * time.Second,
|
|
}
|
|
if opts.PprofAddr != "" {
|
|
ps, perr := StartPprof(opts.PprofAddr, opts.Log)
|
|
if perr != nil {
|
|
s.log.Error("pprof start failed", slog.String("component", "daemon"), slog.Any("err", perr))
|
|
} else {
|
|
s.pprofServer = ps
|
|
}
|
|
}
|
|
return s
|
|
}
|
|
|
|
// Addr returns the configured listen address.
|
|
func (s *Server) Addr() string { return s.addr }
|
|
|
|
// MarkReady flips the readiness flag to true. The /readyz endpoint returns
|
|
// 200 only when this flag is set AND the database is reachable.
|
|
func (s *Server) MarkReady() { s.ready.Store(true) }
|
|
|
|
// MarkNotReady flips the readiness flag to false. Called at shutdown start
|
|
// so load balancers stop routing traffic.
|
|
func (s *Server) MarkNotReady() { s.ready.Store(false) }
|
|
|
|
// Ready reports the current readiness flag.
|
|
func (s *Server) Ready() bool { return s.ready.Load() }
|
|
|
|
// mux builds the route table. Handlers are split across files:
|
|
// - health.go /healthz, /readyz, /v1/status
|
|
// - jobs_handler.go /v1/jobs/*
|
|
// - nodes_handler.go /v1/nodes/*
|
|
// - tasks_handler.go /v1/tasks/*
|
|
// - dispatch_handler.go /orca.v1.Dispatch/* (P02; mounted only if
|
|
// RegisterDispatch was called)
|
|
func (s *Server) mux() http.Handler {
|
|
mux := http.NewServeMux()
|
|
mux.HandleFunc("/healthz", s.handleHealthz)
|
|
mux.HandleFunc("/readyz", s.handleReadyz)
|
|
mux.HandleFunc("/v1/status", s.handleStatus)
|
|
mux.HandleFunc("/v1/jobs", s.handleJobsCollection)
|
|
mux.HandleFunc("/v1/jobs/", s.handleJobsItem)
|
|
mux.HandleFunc("/v1/nodes", s.handleNodesCollection)
|
|
mux.HandleFunc("/v1/tasks", s.handleTasksCollection)
|
|
if s.dispatch != nil {
|
|
s.dispatch.Mount(mux)
|
|
}
|
|
return bodyLimitMiddleware(loggingMiddleware(s.log, mux))
|
|
}
|
|
|
|
// RegisterDispatch attaches the orca.v1.Dispatch service to the
|
|
// daemon. Call before Start(). The dispatch routes are mounted at
|
|
// /orca.v1.Dispatch/Submit and /orca.v1.Dispatch/Status.
|
|
func (s *Server) RegisterDispatch(h *DispatchHandlers) {
|
|
if h == nil {
|
|
return
|
|
}
|
|
s.dispatch = h
|
|
s.log.Info("dispatch handlers registered",
|
|
slog.String("component", "daemon"),
|
|
slog.String("submit", "/orca.v1.Dispatch/Submit"),
|
|
slog.String("status", "/orca.v1.Dispatch/Status"),
|
|
)
|
|
}
|
|
|
|
// Start runs the HTTP server. Returns http.ErrServerClosed on clean shutdown.
|
|
// R-021 / REQ-123: the daemon MUST run in mTLS mode (no plaintext).
|
|
// If StartMTLS has not been called, Start refuses to run.
|
|
func (s *Server) Start() error {
|
|
if s.mtls == nil {
|
|
s.log.Error("daemon refuses to start in plaintext mode",
|
|
slog.String("component", "daemon"),
|
|
slog.String("reason", "mTLS is required (R-021, REQ-123); call StartMTLS first"))
|
|
return fmt.Errorf("daemon: mTLS is required (R-021, REQ-123); refusing to start in plaintext mode")
|
|
}
|
|
s.log.Info("daemon starting (mTLS required)",
|
|
slog.String("addr", s.addr),
|
|
slog.String("component", "daemon"))
|
|
return s.httpServer.ListenAndServe()
|
|
}
|
|
|
|
// Shutdown gracefully stops the server, bounded by ctx. It also flips the
|
|
// readiness flag to false so /readyz returns 503 immediately.
|
|
func (s *Server) Shutdown(ctx context.Context) error {
|
|
s.MarkNotReady()
|
|
s.log.Info("daemon shutting down", slog.String("component", "daemon"))
|
|
if s.pprofServer != nil {
|
|
if perr := s.pprofServer.Shutdown(ctx); perr != nil {
|
|
s.log.Error("pprof shutdown failed", slog.String("component", "daemon"), slog.Any("err", perr))
|
|
}
|
|
}
|
|
return s.httpServer.Shutdown(ctx)
|
|
}
|
|
|
|
// IsShutdownErr reports whether err is the expected error from a stopped server.
|
|
func IsShutdownErr(err error) bool {
|
|
return errors.Is(err, http.ErrServerClosed)
|
|
}
|