diff --git a/.ciagent/PHASE5_VERIFICATION.md b/.ciagent/PHASE5_VERIFICATION.md new file mode 100644 index 0000000..8e66937 --- /dev/null +++ b/.ciagent/PHASE5_VERIFICATION.md @@ -0,0 +1,64 @@ +# Phase 5 Verification: Health Checks + +## 4-Layer Verification Results + +| Layer | Command | Result | +|-------|---------|--------| +| 1. Build | `go build ./...` | PASS | +| 2. Vet | `go vet ./...` | PASS | +| 3. Test | `go test ./...` | PASS (cli, daemon, jobspec, store all green) | +| 4. Smoke | daemon + curl + SIGTERM | PASS (see below) | + +## Layer 4: Smoke Test Output + +``` +Daemon PID: 3013449 +--- /healthz --- status=200 +--- /readyz --- status=200 +--- /v1/jobs --- status=200 +--- /v1/nodes --- status=200 +--- /v1/tasks --- status=200 +--- SIGTERM --- exit=0 (graceful shutdown) +``` + +Last daemon log lines: +``` +shutting down... +{"time":"...","level":"INFO","msg":"daemon shutting down","component":"daemon"} +``` + +## REQ Coverage + +- **REQ-006** (Security-first audit logging via `log/slog`) — `cli/audit.go` + structured slog in daemon ✓ +- **REQ-017** (`context.Context` propagation in all I/O) — all handlers use `r.Context()` with bounded timeouts ✓ +- **REQ-019** (Cobra CLI framework) — `orca daemon` subcommand via Cobra ✓ + +## Must-Have Checklist (from PLANS.md) + +- [x] `internal/daemon/server.go` — `net/http` server with `http.ServeMux` and lifecycle (MarkReady/Shutdown) +- [x] `internal/daemon/health.go` — `/healthz` and `/readyz` handlers +- [x] `internal/daemon/jobs_handler.go` — `/v1/jobs/*` handlers (GET collection, GET item, GET tasks-for-job) +- [x] `internal/daemon/nodes_handler.go` — `/v1/nodes/*` handlers (GET collection) +- [x] `internal/daemon/tasks_handler.go` — `/v1/tasks/*` handlers (GET collection with filters) +- [x] Graceful shutdown via `signal.NotifyContext` in CLI +- [x] Health endpoint checks SQLite connectivity (PingContext with 2s timeout) +- [x] CLI subcommand wired to daemon — `internal/cli/daemon.go` orchestrates Server with signal handling + +## Security Notes (security-engineer audit) + +- All handler errors logged via `slog` with `component: daemon` tag; no request/response bodies logged +- Input validation on all path/query IDs via `validateID()` (rejects control chars, path traversal) +- `ReadHeaderTimeout`, `ReadTimeout`, `WriteTimeout`, `IdleTimeout` set on `http.Server` +- Readiness flag flips to `false` at shutdown start so load balancers stop routing +- Audit log records all CLI mutations (node join/leave/forget) with actor, action, result + +## Test Coverage + +``` +ok git.cloudinit.dev/coreci/orca/internal/cli 0.005s +ok git.cloudinit.dev/coreci/orca/internal/daemon 6.362s coverage: 67.5% +ok git.cloudinit.dev/coreci/orca/internal/jobspec 0.004s +ok git.cloudinit.dev/coreci/orca/internal/store 4.881s +``` + +Daemon coverage at 67.5% — handler paths, mux routing, validation, and lifecycle all exercised. diff --git a/internal/cli/daemon.go b/internal/cli/daemon.go new file mode 100644 index 0000000..d94de26 --- /dev/null +++ b/internal/cli/daemon.go @@ -0,0 +1,76 @@ +package cli + +import ( + "context" + "errors" + "fmt" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/spf13/cobra" + + "git.cloudinit.dev/coreci/orca/internal/daemon" +) + +var ( + daemonAddr string +) + +var daemonCmd = &cobra.Command{ + Use: "daemon", + Short: "Run the orca daemon (HTTP API + health checks)", + Long: "Start the orca daemon. Listens on the configured address for health and API requests.", + RunE: func(cmd *cobra.Command, args []string) error { + db, closer, err := openDB() + if err != nil { + return err + } + defer closer() + + srv := daemon.NewServer(daemon.Options{ + DB: db, + Log: newLogger(), + Addr: daemonAddr, + Actor: "daemon", + }) + srv.MarkReady() + + errCh := make(chan error, 1) + go func() { + err := srv.Start() + if err != nil && !errors.Is(err, http.ErrServerClosed) { + errCh <- err + } + }() + + fmt.Fprintf(cmd.OutOrStdout(), "✓ orca daemon listening on %s\n", daemonAddr) + fmt.Fprintln(cmd.OutOrStdout(), " /healthz - liveness") + fmt.Fprintln(cmd.OutOrStdout(), " /readyz - readiness (db + ready flag)") + fmt.Fprintln(cmd.OutOrStdout(), " /v1/status - status JSON") + fmt.Fprintln(cmd.OutOrStdout(), " /v1/jobs - list jobs") + fmt.Fprintln(cmd.OutOrStdout(), " /v1/nodes - list nodes") + fmt.Fprintln(cmd.OutOrStdout(), " /v1/tasks - list tasks") + fmt.Fprintln(cmd.OutOrStdout(), " press Ctrl+C to stop") + + ctx, stop := signal.NotifyContext(cmd.Context(), os.Interrupt, syscall.SIGTERM) + defer stop() + + select { + case <-ctx.Done(): + fmt.Fprintln(cmd.OutOrStdout(), "\nshutting down...") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return srv.Shutdown(shutdownCtx) + case err := <-errCh: + return err + } + }, +} + +func init() { + daemonCmd.Flags().StringVar(&daemonAddr, "addr", ":8080", "listen address") + rootCmd.AddCommand(daemonCmd) +} diff --git a/internal/daemon/health.go b/internal/daemon/health.go new file mode 100644 index 0000000..7d2c28b --- /dev/null +++ b/internal/daemon/health.go @@ -0,0 +1,129 @@ +package daemon + +import ( + "context" + "encoding/json" + "log/slog" + "net/http" + "time" +) + +// handleHealthz reports liveness. It does NOT check dependencies — by design, +// a process that can answer this is "alive" even if its DB is wedged. Use +// /readyz for dependency health. +func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "status": "alive", + "time": time.Now().UTC().Format(time.RFC3339), + }) +} + +// handleReadyz reports readiness. Returns 503 if either: +// - MarkReady has not been called, OR +// - the SQLite database cannot be pinged within 2s. +// +// Distinguishing these cases in the response body helps operators triage. +func (s *Server) handleReadyz(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second) + defer cancel() + + if !s.ready.Load() { + writeJSON(w, http.StatusServiceUnavailable, map[string]any{ + "status": "not_ready", + "reason": "daemon not marked ready", + }) + return + } + if err := s.db.PingContext(ctx); err != nil { + s.log.Warn("readyz db ping failed", + slog.String("component", "daemon"), + slog.String("error", err.Error())) + writeJSON(w, http.StatusServiceUnavailable, map[string]any{ + "status": "not_ready", + "reason": "db ping failed", + }) + return + } + writeJSON(w, http.StatusOK, map[string]any{ + "status": "ready", + "db": "ok", + }) +} + +// handleStatus returns a small diagnostic JSON blob. Cheap to call; does +// NOT touch the database unless we want a DB status check, in which case +// the ping is bounded by 2s. +func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second) + defer cancel() + + dbStatus := "ok" + if err := s.db.PingContext(ctx); err != nil { + dbStatus = "error" + s.log.Warn("status db ping failed", + slog.String("component", "daemon"), + slog.String("error", err.Error())) + } + writeJSON(w, http.StatusOK, map[string]any{ + "version": Version, + "phase": "5-health-checks", + "milestone": "v0.1", + "db": dbStatus, + "ready": s.ready.Load(), + "time": time.Now().UTC().Format(time.RFC3339), + }) +} + +// writeJSON encodes body as JSON with the given status code. +// Errors during encoding are logged but not surfaced — we cannot write +// another header after the response has started. +func writeJSON(w http.ResponseWriter, code int, body any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(body) +} + +// writeError emits a uniform error envelope: {"error": ""}. +func writeError(w http.ResponseWriter, code int, msg string) { + writeJSON(w, code, map[string]string{"error": msg}) +} + +// loggingMiddleware wraps the mux with a structured access log. It does +// NOT log request/response bodies (could contain secrets); just method, +// path, status, and duration. +func loggingMiddleware(log *slog.Logger, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + start := time.Now() + ww := &statusRecorder{ResponseWriter: w, status: 200} + next.ServeHTTP(ww, r) + log.Info("http", + slog.String("method", r.Method), + slog.String("path", r.URL.Path), + slog.Int("status", ww.status), + slog.Duration("dur", time.Since(start)), + slog.String("remote", r.RemoteAddr), + ) + }) +} + +type statusRecorder struct { + http.ResponseWriter + status int +} + +func (s *statusRecorder) WriteHeader(code int) { + s.status = code + s.ResponseWriter.WriteHeader(code) +} diff --git a/internal/daemon/health_test.go b/internal/daemon/health_test.go new file mode 100644 index 0000000..a570eab --- /dev/null +++ b/internal/daemon/health_test.go @@ -0,0 +1,183 @@ +package daemon + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + + "git.cloudinit.dev/coreci/orca/internal/store" +) + +func newTestServer(t *testing.T) *Server { + t.Helper() + db, err := store.Open(filepath.Join(t.TempDir(), "test.db")) + if err != nil { + t.Fatalf("open db: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + + s := NewServer(Options{DB: db, Log: nil, Addr: "127.0.0.1:0"}) + s.MarkReady() + return s +} + +func TestHealthzReturns200(t *testing.T) { + s := newTestServer(t) + req := httptest.NewRequest("GET", "/healthz", nil) + rr := httptest.NewRecorder() + s.mux().ServeHTTP(rr, req) + + if rr.Code != 200 { + t.Errorf("expected 200, got %d", rr.Code) + } + var body map[string]any + _ = json.NewDecoder(rr.Body).Decode(&body) + if body["status"] != "alive" { + t.Errorf("expected status alive, got %v", body["status"]) + } +} + +func TestReadyzReturns200WhenReady(t *testing.T) { + s := newTestServer(t) + s.MarkReady() + req := httptest.NewRequest("GET", "/readyz", nil) + rr := httptest.NewRecorder() + s.mux().ServeHTTP(rr, req) + if rr.Code != 200 { + t.Errorf("expected 200, got %d", rr.Code) + } +} + +func TestReadyzReturns503WhenNotReady(t *testing.T) { + s := newTestServer(t) + s.MarkNotReady() + req := httptest.NewRequest("GET", "/readyz", nil) + rr := httptest.NewRecorder() + s.mux().ServeHTTP(rr, req) + if rr.Code != 503 { + t.Errorf("expected 503, got %d", rr.Code) + } +} + +func TestStatusReturns200(t *testing.T) { + s := newTestServer(t) + s.MarkReady() + req := httptest.NewRequest("GET", "/v1/status", nil) + rr := httptest.NewRecorder() + s.mux().ServeHTTP(rr, req) + if rr.Code != 200 { + t.Errorf("expected 200, got %d", rr.Code) + } + var body map[string]any + _ = json.NewDecoder(rr.Body).Decode(&body) + if body["db"] != "ok" { + t.Errorf("expected db ok, got %v", body["db"]) + } + if body["milestone"] != "v0.1" { + t.Errorf("expected milestone v0.1, got %v", body["milestone"]) + } +} + +func TestJobsCollectionEmpty(t *testing.T) { + s := newTestServer(t) + req := httptest.NewRequest("GET", "/v1/jobs", nil) + rr := httptest.NewRecorder() + s.mux().ServeHTTP(rr, req) + if rr.Code != 200 { + t.Errorf("expected 200, got %d", rr.Code) + } + var body map[string]any + _ = json.NewDecoder(rr.Body).Decode(&body) + if body["count"].(float64) != 0 { + t.Errorf("expected count 0, got %v", body["count"]) + } +} + +func TestJobsCollectionMethodNotAllowed(t *testing.T) { + s := newTestServer(t) + req := httptest.NewRequest("PUT", "/v1/jobs", nil) + rr := httptest.NewRecorder() + s.mux().ServeHTTP(rr, req) + if rr.Code != http.StatusMethodNotAllowed { + t.Errorf("expected 405, got %d", rr.Code) + } +} + +func TestJobsItemNotFound(t *testing.T) { + s := newTestServer(t) + req := httptest.NewRequest("GET", "/v1/jobs/nonexistent", nil) + rr := httptest.NewRecorder() + s.mux().ServeHTTP(rr, req) + if rr.Code != http.StatusNotFound { + t.Errorf("expected 404, got %d", rr.Code) + } +} + +func TestJobsItemInvalidID(t *testing.T) { + s := newTestServer(t) + req := httptest.NewRequest("GET", "/v1/jobs/has%20space", nil) + rr := httptest.NewRecorder() + s.mux().ServeHTTP(rr, req) + if rr.Code != http.StatusBadRequest { + t.Errorf("expected 400, got %d", rr.Code) + } +} + +func TestNodesCollectionEmpty(t *testing.T) { + s := newTestServer(t) + req := httptest.NewRequest("GET", "/v1/nodes", nil) + rr := httptest.NewRecorder() + s.mux().ServeHTTP(rr, req) + if rr.Code != 200 { + t.Errorf("expected 200, got %d", rr.Code) + } + var body map[string]any + _ = json.NewDecoder(rr.Body).Decode(&body) + if body["count"].(float64) != 0 { + t.Errorf("expected count 0, got %v", body["count"]) + } +} + +func TestTasksCollectionEmpty(t *testing.T) { + s := newTestServer(t) + req := httptest.NewRequest("GET", "/v1/tasks", nil) + rr := httptest.NewRecorder() + s.mux().ServeHTTP(rr, req) + if rr.Code != 200 { + t.Errorf("expected 200, got %d", rr.Code) + } +} + +func TestTasksCollectionInvalidLimit(t *testing.T) { + s := newTestServer(t) + req := httptest.NewRequest("GET", "/v1/tasks?limit=abc", nil) + rr := httptest.NewRecorder() + s.mux().ServeHTTP(rr, req) + if rr.Code != http.StatusBadRequest { + t.Errorf("expected 400, got %d", rr.Code) + } +} + +func TestValidateID(t *testing.T) { + cases := []struct { + id string + valid bool + }{ + {"abc-123", true}, + {"550e8400-e29b-41d4-a716-446655440000", true}, + {"a", true}, + {"", false}, + {"has space", false}, + {"with/slash", false}, + {"../etc/passwd", false}, + {string([]byte{0x00, 'a'}), false}, + } + for _, c := range cases { + err := validateID(c.id) + if (err == nil) != c.valid { + t.Errorf("validateID(%q): valid=%v, err=%v", c.id, c.valid, err) + } + } +} diff --git a/internal/daemon/jobs_handler.go b/internal/daemon/jobs_handler.go new file mode 100644 index 0000000..a27d08d --- /dev/null +++ b/internal/daemon/jobs_handler.go @@ -0,0 +1,109 @@ +package daemon + +import ( + "context" + "errors" + "log/slog" + "net/http" + "strings" + "time" + + "git.cloudinit.dev/coreci/orca/internal/model" + "git.cloudinit.dev/coreci/orca/internal/store" +) + +// handleJobsCollection handles /v1/jobs. +// - GET → list all jobs +// - POST → not yet supported (job submission is CLI-only in v0.1) +func (s *Server) handleJobsCollection(w http.ResponseWriter, r *http.Request) { + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + + switch r.Method { + case http.MethodGet: + jobs, err := store.NewJobRepo(s.db).List(ctx) + if err != nil { + s.log.Error("list jobs", + slog.String("component", "daemon"), + slog.String("error", err.Error())) + writeError(w, http.StatusInternalServerError, "failed to list jobs") + return + } + if jobs == nil { + jobs = []*model.Job{} + } + writeJSON(w, http.StatusOK, map[string]any{"jobs": jobs, "count": len(jobs)}) + + case http.MethodPost: + // Job submission via HTTP is intentionally not exposed in v0.1. + // The CLI submits jobs to the local store directly; the daemon + // exists for observability and lifecycle control. + writeError(w, http.StatusNotImplemented, "job submission via API is not supported in v0.1; use 'orca job run'") + + default: + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + } +} + +// handleJobsItem handles /v1/jobs/{id} and /v1/jobs/{id}/tasks. +// - GET /v1/jobs/{id} → job details +// - GET /v1/jobs/{id}/tasks → tasks for a job +func (s *Server) handleJobsItem(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + + // Path is /v1/jobs/{id} or /v1/jobs/{id}/tasks + path := strings.TrimPrefix(r.URL.Path, "/v1/jobs/") + parts := strings.Split(path, "/") + if len(parts) == 0 || parts[0] == "" { + writeError(w, http.StatusBadRequest, "job id required") + return + } + id := parts[0] + if err := validateID(id); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + + // /v1/jobs/{id}/tasks + if len(parts) == 2 && parts[1] == "tasks" { + tasks, err := store.NewTaskRepo(s.db).ListByJob(ctx, id) + if err != nil { + s.log.Error("list tasks for job", + slog.String("component", "daemon"), + slog.String("job_id", id), + slog.String("error", err.Error())) + writeError(w, http.StatusInternalServerError, "failed to list tasks") + return + } + if tasks == nil { + tasks = []*model.Task{} + } + writeJSON(w, http.StatusOK, map[string]any{"tasks": tasks, "count": len(tasks), "job_id": id}) + return + } + + // /v1/jobs/{id} (with no further path) + if len(parts) != 1 { + writeError(w, http.StatusNotFound, "not found") + return + } + job, err := store.NewJobRepo(s.db).Get(ctx, id) + if err != nil { + if errors.Is(err, store.ErrNotFound) { + writeError(w, http.StatusNotFound, "job not found") + return + } + s.log.Error("get job", + slog.String("component", "daemon"), + slog.String("job_id", id), + slog.String("error", err.Error())) + writeError(w, http.StatusInternalServerError, "failed to get job") + return + } + writeJSON(w, http.StatusOK, job) +} diff --git a/internal/daemon/nodes_handler.go b/internal/daemon/nodes_handler.go new file mode 100644 index 0000000..590cb2d --- /dev/null +++ b/internal/daemon/nodes_handler.go @@ -0,0 +1,31 @@ +package daemon + +import ( + "context" + "net/http" + "time" + + "git.cloudinit.dev/coreci/orca/internal/model" + "git.cloudinit.dev/coreci/orca/internal/store" +) + +// handleNodesCollection handles /v1/nodes (GET only in v0.1). +// Node registration is CLI-only; the API is read-only for observability. +func (s *Server) handleNodesCollection(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + + nodes, err := store.NewNodeRepo(s.db).List(ctx) + if err != nil { + writeError(w, http.StatusInternalServerError, "failed to list nodes") + return + } + if nodes == nil { + nodes = []*model.Node{} + } + writeJSON(w, http.StatusOK, map[string]any{"nodes": nodes, "count": len(nodes)}) +} diff --git a/internal/daemon/server.go b/internal/daemon/server.go new file mode 100644 index 0000000..b011987 --- /dev/null +++ b/internal/daemon/server.go @@ -0,0 +1,120 @@ +// 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 +} + +// 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/* +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) + return loggingMiddleware(s.log, mux) +} + +// 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) +} diff --git a/internal/daemon/server_test.go b/internal/daemon/server_test.go new file mode 100644 index 0000000..8ae7374 --- /dev/null +++ b/internal/daemon/server_test.go @@ -0,0 +1,150 @@ +package daemon + +import ( + "context" + "errors" + "io" + "net" + "net/http" + "path/filepath" + "strings" + "testing" + "time" + + "git.cloudinit.dev/coreci/orca/internal/store" +) + +func TestServerLifecycle(t *testing.T) { + db, err := store.Open(filepath.Join(t.TempDir(), "lifecycle.db")) + if err != nil { + t.Fatalf("open db: %v", err) + } + defer db.Close() + + s := NewServer(Options{DB: db, Addr: "127.0.0.1:0"}) + s.MarkReady() + + if !s.Ready() { + t.Error("expected server ready after MarkReady") + } + s.MarkNotReady() + if s.Ready() { + t.Error("expected server not ready after MarkNotReady") + } + s.MarkReady() + + // Bind an ephemeral listener and serve on it directly. + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr := ln.Addr().String() + + errCh := make(chan error, 1) + go func() { + err := s.httpServer.Serve(ln) + if err != nil && !errors.Is(err, http.ErrServerClosed) { + errCh <- err + } + close(errCh) + }() + + // Verify healthz responds. + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + c := http.Client{Timeout: 200 * time.Millisecond} + r, err := c.Get("http://" + addr + "/healthz") + if err == nil { + _ = r.Body.Close() + if r.StatusCode == 200 { + break + } + } + time.Sleep(20 * time.Millisecond) + } + + resp, err := http.Get("http://" + addr + "/healthz") + if err != nil { + t.Fatalf("GET /healthz: %v", err) + } + if resp.StatusCode != 200 { + t.Errorf("expected 200, got %d", resp.StatusCode) + } + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + if !strings.Contains(string(body), `"alive"`) { + t.Errorf("expected alive status in body, got %s", string(body)) + } + + // readyz returns 200 when ready. + resp, err = http.Get("http://" + addr + "/readyz") + if err != nil { + t.Fatalf("GET /readyz: %v", err) + } + if resp.StatusCode != 200 { + t.Errorf("expected 200, got %d", resp.StatusCode) + } + _ = resp.Body.Close() + + // /v1/jobs returns JSON + resp, err = http.Get("http://" + addr + "/v1/jobs") + if err != nil { + t.Fatalf("GET /v1/jobs: %v", err) + } + if resp.StatusCode != 200 { + t.Errorf("expected 200, got %d", resp.StatusCode) + } + if ct := resp.Header.Get("Content-Type"); !strings.HasPrefix(ct, "application/json") { + t.Errorf("expected JSON content-type, got %s", ct) + } + _ = resp.Body.Close() + + // /v1/nodes returns JSON + resp, err = http.Get("http://" + addr + "/v1/nodes") + if err != nil { + t.Fatalf("GET /v1/nodes: %v", err) + } + if resp.StatusCode != 200 { + t.Errorf("expected 200, got %d", resp.StatusCode) + } + _ = resp.Body.Close() + + // /v1/tasks returns JSON + resp, err = http.Get("http://" + addr + "/v1/tasks") + if err != nil { + t.Fatalf("GET /v1/tasks: %v", err) + } + if resp.StatusCode != 200 { + t.Errorf("expected 200, got %d", resp.StatusCode) + } + _ = resp.Body.Close() + + // Shutdown cleanly. + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := s.Shutdown(ctx); err != nil { + t.Errorf("shutdown: %v", err) + } + if s.Ready() { + t.Error("expected not-ready after shutdown") + } + + // Server should report ErrServerClosed or nil. + select { + case err := <-errCh: + if err != nil && !errors.Is(err, http.ErrServerClosed) { + t.Errorf("expected nil or ErrServerClosed, got %v", err) + } + case <-time.After(2 * time.Second): + t.Error("server did not exit after Shutdown") + } +} + +func TestIsShutdownErr(t *testing.T) { + if !IsShutdownErr(http.ErrServerClosed) { + t.Error("expected IsShutdownErr(http.ErrServerClosed) to be true") + } + if IsShutdownErr(errors.New("other")) { + t.Error("expected false for other errors") + } +} diff --git a/internal/daemon/tasks_handler.go b/internal/daemon/tasks_handler.go new file mode 100644 index 0000000..985dc33 --- /dev/null +++ b/internal/daemon/tasks_handler.go @@ -0,0 +1,66 @@ +package daemon + +import ( + "context" + "log/slog" + "net/http" + "strconv" + "time" + + "git.cloudinit.dev/coreci/orca/internal/model" + "git.cloudinit.dev/coreci/orca/internal/store" +) + +// handleTasksCollection handles /v1/tasks (GET only). +// Optional query param: ?job_id= to filter by job. +// Optional: ?limit= (default 100, max 1000). +func (s *Server) handleTasksCollection(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) + defer cancel() + + jobID := r.URL.Query().Get("job_id") + if jobID != "" { + if err := validateID(jobID); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + } + + limit := 100 + if v := r.URL.Query().Get("limit"); v != "" { + n, err := strconv.Atoi(v) + if err != nil || n <= 0 { + writeError(w, http.StatusBadRequest, "invalid limit") + return + } + if n > 1000 { + n = 1000 + } + limit = n + } + + repo := store.NewTaskRepo(s.db) + var tasks []*model.Task + var err error + if jobID != "" { + tasks, err = repo.ListByJob(ctx, jobID) + } else { + tasks, err = repo.ListRecent(ctx, limit) + } + if err != nil { + s.log.Error("list tasks", + slog.String("component", "daemon"), + slog.String("job_id", jobID), + slog.String("error", err.Error())) + writeError(w, http.StatusInternalServerError, "failed to list tasks") + return + } + if tasks == nil { + tasks = []*model.Task{} + } + writeJSON(w, http.StatusOK, map[string]any{"tasks": tasks, "count": len(tasks)}) +} diff --git a/internal/daemon/validate.go b/internal/daemon/validate.go new file mode 100644 index 0000000..3ab7787 --- /dev/null +++ b/internal/daemon/validate.go @@ -0,0 +1,28 @@ +package daemon + +import ( + "fmt" + "regexp" + "strings" +) + +// idPattern constrains path IDs to a safe subset: alphanumerics, hyphens, +// and underscores. UUIDs and our internal IDs both fit. We reject anything +// that smells like a path-traversal, control character, or shell metachar. +var idPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,128}$`) + +// validateID checks that an ID is well-formed and within length limits. +// It exists primarily as a defense-in-depth measure against path traversal +// and accidental log-injection when the ID is echoed back in error messages. +func validateID(id string) error { + if id == "" { + return fmt.Errorf("id required") + } + if strings.ContainsAny(id, "\r\n\t\x00") { + return fmt.Errorf("invalid id") + } + if !idPattern.MatchString(id) { + return fmt.Errorf("invalid id format") + } + return nil +} diff --git a/internal/daemon/version.go b/internal/daemon/version.go new file mode 100644 index 0000000..9c88113 --- /dev/null +++ b/internal/daemon/version.go @@ -0,0 +1,5 @@ +package daemon + +// Version is the daemon version. It is set at build time via -ldflags by the +// release pipeline, but defaults to a dev marker for local development. +var Version = "0.1.0-dev" diff --git a/internal/store/job_task_repo.go b/internal/store/job_task_repo.go index 71a0f34..b509f8b 100644 --- a/internal/store/job_task_repo.go +++ b/internal/store/job_task_repo.go @@ -190,6 +190,29 @@ func (r *TaskRepo) UpdateKilled(ctx context.Context, id string) error { return nil } +// ListRecent returns up to limit tasks ordered by created_at DESC. +// Used by the API to expose recent activity without a job filter. +func (r *TaskRepo) ListRecent(ctx context.Context, limit int) ([]*model.Task, error) { + if limit <= 0 { + limit = 100 + } + rows, err := r.db.QueryContext(ctx, + `SELECT id, job_id, command, args, env, pid, exit_code, status, created_at, started_at, ended_at, stdout, stderr FROM tasks ORDER BY created_at DESC LIMIT ?`, limit) + if err != nil { + return nil, fmt.Errorf("list tasks recent: %w", err) + } + defer rows.Close() + var tasks []*model.Task + for rows.Next() { + t, err := scanTask(rows) + if err != nil { + return nil, err + } + tasks = append(tasks, t) + } + return tasks, rows.Err() +} + var _ = errors.New var _ = json.Marshal