5dba3cef80
Wave B of P02. Wires the data + engine + transport layers into the
daemon HTTP surface and the CLI.
- internal/engine/executor.go — adds Submit(specBytes) and
Status(jobID) entry points to satisfy engine.LocalExecutor
(used by the dispatcher). Submit parses a minimal JSON wire
spec with name/command/args/env fields; Status reads from
store.JobRepo and returns the stringified model.JobStatus.
- internal/engine/dispatcher.go — Dispatcher struct with
LocalExecutor + capacity repo + peer registry + idempotency
dedupe store. Submit(target, spec, idempotencyKey) does the
local-fit-check then bin-packing pick; if no local capacity
and target is empty, falls through to a peer. dispatchTo /
dispatchToPeer open mTLS clients (no cert presented by the
client in P02; the server uses RequireAndVerifyClientCert
but P02 ships with the cert-pool wiring without enforcing
client certs on the dispatch endpoint — P03 hardening).
LocalSubmit/LocalStatus satisfy transport.Dispatcher.
- internal/transport/dispatch.go — SubmitHandler and
StatusHandler (http.Handler). SubmitHandler honors
X-Orca-Idempotency-Key for dedupe replay. Submit/Status
Request/Response wire structs. DispatchClient wraps
mTLS HTTP client with the retry loop. The retry Submit
is implemented as a direct loop (not via Do[T]) because
the response-decode path doesn't fit the generic shape
cleanly.
- internal/daemon/dispatch_handler.go — DispatchHandlers
groups Submit+Status; Mount(mux) attaches both routes.
- internal/daemon/server.go — Server gets a dispatch field;
RegisterDispatch(h) attaches the handlers; mux() mounts
them at /orca.v1.Dispatch/{Submit,Status}.
- internal/daemon/dispatch_test.go — round-trip, idempotency
dedupe, and validation (empty spec=400, GET=405) coverage.
- internal/cli/daemon.go — wires the dispatch service into
the daemon: executor + peer registry + dispatcher +
RegisterDispatch. Adds /orca.v1.Dispatch/* to the startup
banner.
- internal/cli/job.go — adds --target and --idempotency-key
to 'orca job run'; routes through the dispatcher when set.
- internal/cli/node_capacity.go — 'orca node capacity
{show,set,list}' for REQ-028. --set takes --cpu, --memory,
--disk, --node. Positivity check on all three numerics.
All tests pass with -race; gofmt -l . clean; go vet ./...
clean. P02 verification commit follows.
---ci---
project: orca
phase: 9
milestone: v0.2
status: execute
---/ci---
152 lines
4.5 KiB
Go
152 lines
4.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
|
|
package daemon
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"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
|
|
|
|
// 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
|
|
}
|
|
|
|
// 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,
|
|
}
|
|
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 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.
|
|
func (s *Server) Start() error {
|
|
s.log.Info("daemon starting",
|
|
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"))
|
|
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)
|
|
}
|