diff --git a/internal/cli/job.go b/internal/cli/job.go index 8e07fe6..ab84af6 100644 --- a/internal/cli/job.go +++ b/internal/cli/job.go @@ -60,16 +60,21 @@ var jobRunCmd = &cobra.Command{ ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Minute) defer cancel() - exec, closer, err := jobExecutor() - if err != nil { - return err - } - defer closer() - - // If --target or --idempotency-key is set, route through the - // dispatcher (which may land the job locally or on a peer - // based on capacity). - if runTarget != "" || runIDKey != "" { + // v0.13 phase-03 scheduler wiring (REQ-151, C-44): decide + // whether to run locally (dev mode / no remote nodes) or + // remotely (scheduler picks a peer, render systemd, SSH-push). + // The deprecated mTLS Dispatcher path (--idempotency-key) is + // retained only for the dual-write window; the new remote path + // uses the CLI-side scheduler + sshpush. + if runIDKey != "" { + // Legacy --idempotency-key dispatch path (deprecated mTLS + // Dispatcher). Retained for backward compat; routes through + // engine.Dispatcher which is scheduled for removal in v0.10. + exec, closer, err := jobExecutor() + if err != nil { + return err + } + defer closer() db, dbCloser, err := openDB() if err != nil { return err @@ -95,25 +100,70 @@ var jobRunCmd = &cobra.Command{ return nil } - job := &model.Job{ - ID: uuid.NewString(), - Name: spec.Name, - Spec: args[0], - Status: model.JobStatusPending, - } - if err := exec.Run(ctx, job, workloadToTaskSpecs(spec)); err != nil { + res, nodesByHost, err := dispatchDecision(ctx, spec, runTarget) + if err != nil { + logDispatch(nil, err) if jsonOutput { - _ = printJSON(map[string]any{"id": job.ID, "status": "failed", "error": err.Error()}) - return err + _ = printJSON(map[string]any{"status": "failed", "error": err.Error()}) } - fmt.Fprintf(cmd.ErrOrStderr(), "✗ Job %s failed: %v\n", job.ID, err) return err } - if jsonOutput { - return printJSON(map[string]any{"id": job.ID, "name": job.Name, "status": "complete"}) + + switch res.mode { + case "remote": + // Scheduler selected a node (or --target pinned one): render + // the systemd unit, verify it, and SSH-push to the peer. + // C-44: a push failure is an error (no local fallback). + unitPaths, derr := deployRemote(ctx, spec, res, nodesByHost) + logDispatch(res, derr) + if derr != nil { + if jsonOutput { + _ = printJSON(map[string]any{"status": "failed", "node": res.node, "error": derr.Error()}) + } + return derr + } + res.unitPaths = unitPaths + if jsonOutput { + return printJSON(map[string]any{ + "status": "deployed", + "node": res.node, + "alloc_id": res.allocID, + "units": unitPaths, + }) + } + fmt.Fprintf(cmd.OutOrStdout(), "✓ Job deployed to %s: %s (%s)\n", res.node, spec.Name, strings.Join(unitPaths, ", ")) + return nil + + case "local": + // Local exec fallback (dev mode: no remote nodes registered). + exec, closer, err := jobExecutor() + if err != nil { + return err + } + defer closer() + job := &model.Job{ + ID: uuid.NewString(), + Name: spec.Name, + Spec: args[0], + Status: model.JobStatusPending, + } + runErr := exec.Run(ctx, job, workloadToTaskSpecs(spec)) + logDispatch(res, runErr) + if runErr != nil { + if jsonOutput { + _ = printJSON(map[string]any{"id": job.ID, "status": "failed", "error": runErr.Error()}) + return runErr + } + fmt.Fprintf(cmd.ErrOrStderr(), "✗ Job %s failed: %v\n", job.ID, runErr) + return runErr + } + if jsonOutput { + return printJSON(map[string]any{"id": job.ID, "name": job.Name, "status": "complete"}) + } + fmt.Fprintf(cmd.OutOrStdout(), "✓ Job complete: %s (%s)\n", job.ID, job.Name) + return nil } - fmt.Fprintf(cmd.OutOrStdout(), "✓ Job complete: %s (%s)\n", job.ID, job.Name) - return nil + return fmt.Errorf("job run: unknown dispatch mode %q", res.mode) }, } diff --git a/internal/cli/job_dispatch.go b/internal/cli/job_dispatch.go new file mode 100644 index 0000000..fdc7dee --- /dev/null +++ b/internal/cli/job_dispatch.go @@ -0,0 +1,435 @@ +// Package cli: job_dispatch.go wires the v0.9 CLI-side scheduler +// (internal/scheduler), the systemd emitter (internal/emitter), and the +// SSH-push transport (internal/sshpush) into `orca job run` +// (REQ-151, binding condition C-44, v0.13 milestone phase 03). +// +// The dispatch flow (replacing the deprecated mTLS Dispatcher path) is: +// +// 1. Load registered nodes from the orca registry (DB) and project them +// into scheduler.NodeInfo + a hostname->model.Node map for SSH-push. +// 2. If --target is set, pin to that node directly (manual override). +// 3. If no --target and no remote nodes are registered (only localhost +// or none), fall back to local exec (backward compat for dev mode). +// 4. If no --target and remote nodes ARE registered, invoke +// scheduler.Schedule -> pick the best node -> render the systemd unit +// via internal/emitter -> systemd-analyze verify (when available) -> +// SSH-push the unit to the target via internal/sshpush. +// +// C-44 (binding condition): if the scheduler selects a node but the +// SSH-push FAILS, return an error. Do NOT silently fall back to local +// execution. Local fallback is ONLY when len(registeredRemoteNodes)==0. +package cli + +import ( + "context" + "errors" + "fmt" + "log/slog" + "os" + "os/exec" + "strings" + "time" + + "git.cloudinit.dev/coreci/orca/internal/certpaths" + "git.cloudinit.dev/coreci/orca/internal/emitter" + "git.cloudinit.dev/coreci/orca/internal/jobspec" + "git.cloudinit.dev/coreci/orca/internal/model" + "git.cloudinit.dev/coreci/orca/internal/scheduler" + "git.cloudinit.dev/coreci/orca/internal/sshpush" + "git.cloudinit.dev/coreci/orca/internal/store" +) + +// jobDispatchTransport is the SSH-push surface `job run` needs for +// remote deployment. *sshpush.Transport satisfies it; tests substitute +// a mock (same pattern as txn.go / job_verify.go). +type jobDispatchTransport interface { + WriteFile(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) error + Exec(ctx context.Context, peer string, cmd string) ([]byte, error) + Close() error +} + +// jobDispatchTransportOverride is the package-level seam. When non-nil +// it replaces the production transport; tests set it and restore nil. +var jobDispatchTransportOverride jobDispatchTransport + +// jobDispatchTransportFromCtx returns the active SSH-push transport. +// Tests override via jobDispatchTransportOverride; production builds a +// real *sshpush.Transport from the orca SSH key + known_hosts paths. +func jobDispatchTransportFromCtx() (jobDispatchTransport, error) { + if jobDispatchTransportOverride != nil { + return jobDispatchTransportOverride, nil + } + keyPath := certpaths.SSHKeyPath() + khPath := certpaths.KnownHostsPath() + return sshpush.NewTransport(keyPath, khPath), nil +} + +// dispatchResult is the outcome of a `job run` dispatch decision. +type dispatchResult struct { + // mode is "local" (local exec fallback) or "remote" (scheduled + + // SSH-pushed to a peer). + mode string + // node is the hostname of the selected/pinned node (remote only). + node string + // allocID is the scheduler allocation id (remote only). + allocID string + // unitPaths is the list of systemd unit paths written (remote only). + unitPaths []string +} + +// dispatchDecision decides how `job run` should execute the spec: +// +// - "local" -> run via the local executor (dev mode / no remote nodes) +// - "remote" -> render + SSH-push the systemd unit to the chosen node +// +// It loads registered nodes from the DB, projects them into +// scheduler.NodeInfo, and consults the scheduler when no --target is +// set. Returns a dispatchResult describing the chosen path; the caller +// performs the actual execution. +// +// C-44: when remote nodes are registered, a scheduling failure returns +// an error (no local fallback). The local fallback ONLY happens when +// there are zero remote nodes registered (only localhost or none). +func dispatchDecision(ctx context.Context, spec *jobspec.WorkloadSpec, target string) (*dispatchResult, map[string]*model.Node, error) { + if spec == nil { + return nil, nil, errors.New("dispatch: nil spec") + } + db, closer, err := openDB() + if err != nil { + return nil, nil, fmt.Errorf("dispatch: open db: %w", err) + } + defer closer() + + nodeRepo := store.NewNodeRepo(db) + capRepo := store.NewCapacityRepo(db) + nodes, err := nodeRepo.List(ctx) + if err != nil { + return nil, nil, fmt.Errorf("dispatch: list nodes: %w", err) + } + caps, err := capRepo.List(ctx) + if err != nil { + return nil, nil, fmt.Errorf("dispatch: list capacity: %w", err) + } + capByNode := make(map[string]*store.NodeCapacity, len(caps)) + for _, c := range caps { + capByNode[c.NodeID] = c + } + + // Project registered nodes into scheduler.NodeInfo. A node counts + // as a "remote" scheduling candidate when it is ready and is NOT + // the localhost node (kind=localhost). localhost is excluded from + // the candidate set so the scheduler only considers real peers; + // when the candidate set is empty we fall back to local exec. + var candidates []scheduler.NodeInfo + remoteNodes := make(map[string]*model.Node) // hostname -> node + for _, n := range nodes { + if n.State != model.NodeStateReady { + continue + } + if n.Kind == string(model.NodeKindLocalhost) { + continue + } + ni := nodeToNodeInfo(n, capByNode[n.ID]) + candidates = append(candidates, ni) + remoteNodes[ni.Hostname] = n + } + + // --target override: pin to the named node. The target may be a + // node ID, name, or hostname. We resolve it against the registered + // nodes (including localhost when explicitly targeted). + if strings.TrimSpace(target) != "" { + chosen, err := resolveTargetNode(ctx, nodeRepo, target) + if err != nil { + return nil, nil, err + } + hostname := chosen.Name + if hostname == "" { + hostname = chosen.ID + } + // Even a localhost target goes through the remote push path + // when explicitly pinned (the operator asked for it). + remoteNodes[hostname] = chosen + return &dispatchResult{ + mode: "remote", + node: hostname, + allocID: allocIDFor(spec, 0), + }, remoteNodes, nil + } + + // No remote nodes registered -> local exec fallback (dev mode). + if len(candidates) == 0 { + return &dispatchResult{mode: "local"}, remoteNodes, nil + } + + // Remote nodes registered -> invoke the scheduler. A scheduling + // failure is an error (C-44: no silent local fallback). + placements, err := scheduler.Schedule(candidates, scheduler.WorkloadRequest{ + Spec: spec, + Namespace: "default", + }) + if err != nil { + return nil, nil, fmt.Errorf("dispatch: schedule: %w", err) + } + if len(placements) == 0 { + return nil, nil, fmt.Errorf("dispatch: scheduler returned no placements for %q", spec.Name) + } + // Job/DaemonSet produce one-or-many placements; for `job run` we + // deploy the first placement (the best-fit node). Multi-replica + // Service fan-out is handled by the txn/apply path, not job run. + p := placements[0] + return &dispatchResult{ + mode: "remote", + node: p.Node, + allocID: p.AllocID, + }, remoteNodes, nil +} + +// deployRemote renders the systemd unit for the spec on the chosen +// node, runs systemd-analyze verify (when available), and SSH-pushes +// the unit files to the peer. Returns the list of unit paths written. +// +// C-44: any render/verify/push failure is returned as an error; the +// caller must NOT fall back to local exec. +func deployRemote(ctx context.Context, spec *jobspec.WorkloadSpec, res *dispatchResult, nodesByHost map[string]*model.Node) ([]string, error) { + if res == nil || res.mode != "remote" { + return nil, errors.New("deployRemote: not a remote dispatch") + } + node, ok := nodesByHost[res.node] + if !ok { + return nil, fmt.Errorf("deployRemote: selected node %q not found in registry", res.node) + } + + // Render the systemd unit via the emitter. The runtime is required + // for the process emitter; a spec with no runtime has nothing to + // ExecStart and is rejected by the emitter. + em := emitter.SystemdEmitter{} + enode := &emitter.Node{ + Hostname: node.Name, + Runtime: []string{"process"}, + Tags: nil, + } + // Advertise the node kind as a runtime so the emitter can branch + // (proxmox nodes expose pve-* runtimes). For process workloads + // this is informational. + if node.Kind == string(model.NodeKindProxmox) { + enode.Runtime = append(enode.Runtime, "proxmox") + } + files, err := em.Render(spec, enode) + if err != nil { + return nil, fmt.Errorf("deployRemote: render unit: %w", err) + } + + // T9: systemd-analyze verify on the rendered unit before deploy. + // Run it locally (the unit is a portable text file); if + // systemd-analyze is not installed, skip silently (dev boxes + // without systemd). A verification FAILURE is an error. + for _, f := range files { + if err := verifySystemdUnit(ctx, f.Path, f.Content); err != nil { + return nil, fmt.Errorf("deployRemote: systemd-analyze verify %s: %w", f.Path, err) + } + } + + // SSH-push the unit files to the peer. + peer := sshPeerFor(node) + transport, err := jobDispatchTransportFromCtx() + if err != nil { + return nil, fmt.Errorf("deployRemote: transport: %w", err) + } + defer transport.Close() + + var written []string + for _, f := range files { + mode := os.FileMode(0o644) + if f.Mode != "" { + // f.Mode is an octal string like "0644". + var m uint64 + if _, perr := fmt.Sscanf(f.Mode, "%o", &m); perr == nil { + mode = os.FileMode(m) + } + } + if err := transport.WriteFile(ctx, peer, f.Path, []byte(f.Content), mode); err != nil { + // C-44: SSH-push failure -> error, NOT local fallback. + return nil, fmt.Errorf("deployRemote: push %s to %s (%s): %w", f.Path, res.node, peer, err) + } + written = append(written, f.Path) + } + + // Reload systemd + enable the unit so it starts at boot. These are + // best-effort; a failure here is surfaced but does not undo the + // push (the unit is on disk). We use systemctl daemon-reload + + // enable --now for each .service unit (.target units for task + // groups are also enabled). + for _, p := range written { + if !strings.HasSuffix(p, ".service") && !strings.HasSuffix(p, ".target") { + continue + } + if _, err := transport.Exec(ctx, peer, fmt.Sprintf("systemctl daemon-reload && systemctl enable --now %s", shellQuoteSystemd(p))); err != nil { + return written, fmt.Errorf("deployRemote: enable %s on %s: %w", p, res.node, err) + } + } + + return written, nil +} + +// verifySystemdUnit runs `systemd-analyze verify` on the rendered unit +// content. The unit is written to a temp file (with its real basename) +// so systemd-analyze resolves fragment paths correctly. When +// systemd-analyze is not on PATH, the check is skipped (dev boxes +// without systemd). A non-zero exit from systemd-analyze is an error. +func verifySystemdUnit(ctx context.Context, unitPath, content string) error { + bin, err := exec.LookPath("systemd-analyze") + if err != nil { + // systemd-analyze not available (e.g. macOS dev box, minimal + // container). Skip verification rather than failing — the + // render layer already validates the spec shape. + return nil + } + base := unitPath + if idx := strings.LastIndex(unitPath, "/"); idx >= 0 { + base = unitPath[idx+1:] + } + // os.CreateTemp appends a random suffix that would strip the + // .service/.target extension systemd-analyze needs to recognize the + // unit. Create the temp file in a dedicated temp dir with the exact + // basename so the extension is preserved. + tmpDir, err := os.MkdirTemp("", "orca-verify-") + if err != nil { + return fmt.Errorf("temp dir: %w", err) + } + defer os.RemoveAll(tmpDir) + tmpPath := tmpDir + "/" + base + if err := os.WriteFile(tmpPath, []byte(content), 0o644); err != nil { + return fmt.Errorf("write temp unit: %w", err) + } + + vctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + cmd := exec.CommandContext(vctx, bin, "verify", tmpPath) + out, err := cmd.CombinedOutput() + if err != nil { + // Trim the temp path from the output so the error reads with + // the real unit path. + msg := strings.TrimSpace(string(out)) + msg = strings.ReplaceAll(msg, tmpPath, unitPath) + return fmt.Errorf("systemd-analyze verify failed: %s", msg) + } + return nil +} + +// nodeToNodeInfo projects a registered model.Node (+ its capacity +// declaration) into a scheduler.NodeInfo. Runtimes are derived from the +// node kind (proxmox -> "proxmox"; else "process"). Tags are sourced +// from node metadata["tags"] (comma-separated) when present. Capacity +// is sourced from the NodeCapacity row when present (else zero, which +// the scheduler treats as always-fits on the capacity axis). +func nodeToNodeInfo(n *model.Node, cap *store.NodeCapacity) scheduler.NodeInfo { + ni := scheduler.NodeInfo{ + Hostname: n.Name, + Kind: n.Kind, + } + if ni.Kind == "" { + ni.Kind = string(model.NodeKindLinux) + } + switch n.Kind { + case string(model.NodeKindProxmox): + ni.Runtimes = []string{"process", "proxmox"} + default: + ni.Runtimes = []string{"process"} + } + if tags := nodeMetadataTag(n, "tags"); tags != "" { + for _, t := range strings.Split(tags, ",") { + t = strings.TrimSpace(t) + if t != "" { + ni.Tags = append(ni.Tags, t) + } + } + } + if cap != nil { + ni.CPU = cap.CPUMillicores + ni.Memory = cap.MemoryMiB + ni.FreeCPU = cap.CPUMillicores + ni.FreeMem = cap.MemoryMiB + } + return ni +} + +// nodeMetadataTag reads a key from the node's metadata map. Returns "" +// when the metadata is nil or the key is absent. +func nodeMetadataTag(n *model.Node, key string) string { + if n == nil || n.Metadata == nil { + return "" + } + return n.Metadata[key] +} + +// resolveTargetNode resolves a --target value (node ID, name, or +// hostname) to a registered *model.Node. Returns an error when the +// target is not found. +func resolveTargetNode(ctx context.Context, repo *store.NodeRepo, target string) (*model.Node, error) { + target = strings.TrimSpace(target) + if target == "" { + return nil, errors.New("resolveTargetNode: empty target") + } + // Try by ID first. + if n, err := repo.Get(ctx, target); err == nil { + return n, nil + } + // Then by name. + if n, err := repo.GetByName(ctx, target); err == nil { + return n, nil + } + return nil, fmt.Errorf("resolveTargetNode: target node %q not found in registry", target) +} + +// sshPeerFor returns the host:port SSH peer address for a node. The +// node's orca Address is the mTLS daemon port (host:8443); SSH uses a +// different port. We derive the host from the orca Address and use the +// SSH port from node metadata["ssh_port"] when present, else 22. +func sshPeerFor(n *model.Node) string { + host := n.Address + if idx := strings.LastIndex(host, ":"); idx >= 0 { + host = host[:idx] + } + // Strip an ipv6 bracket if present. + host = strings.TrimPrefix(host, "[") + host = strings.TrimSuffix(host, "]") + port := "22" + if n != nil && n.Metadata != nil { + if p, ok := n.Metadata["ssh_port"]; ok && strings.TrimSpace(p) != "" { + port = strings.TrimSpace(p) + } + } + return host + ":" + port +} + +// allocIDFor renders a stable allocation id for a spec index, matching +// the scheduler's allocID format (ns/name-idx). +func allocIDFor(spec *jobspec.WorkloadSpec, idx int) string { + return fmt.Sprintf("default/%s-%d", spec.Name, idx) +} + +// shellQuoteSystemd single-quotes a path for safe shell interpolation +// in the remote systemctl command. Mirrors sshpush.shellQuote. +func shellQuoteSystemd(s string) string { + return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'" +} + +// logDispatch records the dispatch decision to the structured logger. +func logDispatch(res *dispatchResult, err error) { + log := slog.Default() + if res == nil { + log.Info("job.dispatch", slog.String("event", "job.dispatch"), slog.String("mode", "error"), slog.Any("error", err)) + return + } + attrs := []any{slog.String("event", "job.dispatch"), slog.String("mode", res.mode)} + if res.node != "" { + attrs = append(attrs, slog.String("node", res.node)) + } + if res.allocID != "" { + attrs = append(attrs, slog.String("alloc_id", res.allocID)) + } + if err != nil { + attrs = append(attrs, slog.Any("error", err)) + } + log.Info("job.dispatch", attrs...) +} diff --git a/internal/cli/job_dispatch_test.go b/internal/cli/job_dispatch_test.go new file mode 100644 index 0000000..9522ebd --- /dev/null +++ b/internal/cli/job_dispatch_test.go @@ -0,0 +1,425 @@ +package cli + +import ( + "bytes" + "context" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "git.cloudinit.dev/coreci/orca/internal/certpaths" + "git.cloudinit.dev/coreci/orca/internal/jobspec" + "git.cloudinit.dev/coreci/orca/internal/model" + "git.cloudinit.dev/coreci/orca/internal/store" +) + +// mockDispatchTransport is a test double for jobDispatchTransport. It +// records calls and returns configured errors. The zero value succeeds +// for every call. +type mockDispatchTransport struct { + mu sync.Mutex + writeCalls []mockDispatchWriteCall + execCalls []mockDispatchExecCall + writeErr error // returned by WriteFile (simulates C-44 push failure) + execErr error + closeCalled bool +} + +type mockDispatchWriteCall struct { + Peer string + Path string + Content string + Mode os.FileMode +} + +type mockDispatchExecCall struct { + Peer string + Cmd string +} + +func (m *mockDispatchTransport) WriteFile(ctx context.Context, peer, path string, content []byte, mode os.FileMode) error { + m.mu.Lock() + defer m.mu.Unlock() + m.writeCalls = append(m.writeCalls, mockDispatchWriteCall{Peer: peer, Path: path, Content: string(content), Mode: mode}) + return m.writeErr +} + +func (m *mockDispatchTransport) Exec(ctx context.Context, peer, cmd string) ([]byte, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.execCalls = append(m.execCalls, mockDispatchExecCall{Peer: peer, Cmd: cmd}) + return nil, m.execErr +} + +func (m *mockDispatchTransport) Close() error { + m.mu.Lock() + defer m.mu.Unlock() + m.closeCalled = true + return nil +} + +// insertRemoteNode registers a ready remote (non-localhost) node in the +// test DB so the scheduler sees it as a candidate. +func insertRemoteNode(t *testing.T, name, addr string) { + t.Helper() + db, err := store.Open(certpaths.DBPath()) + if err != nil { + t.Fatalf("open db: %v", err) + } + defer db.Close() + repo := store.NewNodeRepo(db) + if err := repo.Insert(context.Background(), &model.Node{ + ID: name, + Name: name, + Address: addr, + State: model.NodeStateReady, + JoinedAt: time.Now().UTC(), + LastSeen: time.Now().UTC(), + Kind: string(model.NodeKindLinux), + OS: "linux", + }); err != nil { + t.Fatalf("insert node %s: %v", name, err) + } +} + +// writeJobMDSpec writes a Markdown jobspec to a temp file and returns +// the path. +func writeJobMDSpec(t *testing.T, content string) string { + t.Helper() + dir := t.TempDir() + p := filepath.Join(dir, "spec.md") + if err := os.WriteFile(p, []byte(content), 0o644); err != nil { + t.Fatalf("write spec: %v", err) + } + return p +} + +const mdJobTrue = "---\n" + + "kind: Job\n" + + "name: true-job\n" + + "runtime:\n" + + " one_of: process\n" + + " command: /bin/true\n" + + "---\n# True\n\nRuns /bin/true.\n" + +// TestREQ151_LocalFallbackNoRemoteNodes (T13): `job run` with no remote +// nodes registered (only localhost or none) runs locally via the +// executor. The output says "Job complete" (local), not "deployed". +func TestREQ151_LocalFallbackNoRemoteNodes(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + resetRootFlags(t) + spec := writeJobMDSpec(t, mdJobTrue) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"job", "run", spec}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("job run local fallback: %v\n%s", err, buf.String()) + } + out := buf.String() + if !strings.Contains(out, "Job complete") { + t.Errorf("expected local 'Job complete' output, got: %s", out) + } + if strings.Contains(out, "deployed") { + t.Errorf("did not expect 'deployed' for local fallback, got: %s", out) + } +} + +// TestREQ151_RemoteNodeScheduledAndPushed (T7): `job run` with a remote +// node registered invokes the scheduler and SSH-pushes the unit. The +// mock transport records the write and the output says "deployed". +func TestREQ151_RemoteNodeScheduledAndPushed(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + insertRemoteNode(t, "worker-1", "10.0.0.5:8443") + resetRootFlags(t) + + mock := &mockDispatchTransport{} + prev := jobDispatchTransportOverride + jobDispatchTransportOverride = mock + defer func() { jobDispatchTransportOverride = prev }() + + spec := writeJobMDSpec(t, mdJobTrue) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"job", "run", spec}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("job run remote: %v\n%s", err, buf.String()) + } + out := buf.String() + if !strings.Contains(out, "deployed to worker-1") { + t.Errorf("expected 'deployed to worker-1', got: %s", out) + } + if len(mock.writeCalls) == 0 { + t.Errorf("expected SSH-push write calls, got 0") + } + // The unit path should be the orca-v1 systemd unit. + wrote := false + for _, c := range mock.writeCalls { + if strings.HasSuffix(c.Path, "orca-v1-true-job.service") { + wrote = true + if !strings.Contains(c.Content, "ExecStart=/bin/true") { + t.Errorf("unit content missing ExecStart:\n%s", c.Content) + } + } + } + if !wrote { + t.Errorf("no write to orca-v1-true-job.service; calls=%+v", mock.writeCalls) + } +} + +// TestREQ151_C44_PushFailureReturnsError (T14, binding condition +// C-44): when the scheduler selects a remote node but SSH-push fails, +// `job run` returns an error. It does NOT silently fall back to local +// execution. +func TestREQ151_C44_PushFailureReturnsError(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + insertRemoteNode(t, "worker-1", "10.0.0.5:8443") + resetRootFlags(t) + + mock := &mockDispatchTransport{writeErr: errMockPush} + prev := jobDispatchTransportOverride + jobDispatchTransportOverride = mock + defer func() { jobDispatchTransportOverride = prev }() + + spec := writeJobMDSpec(t, mdJobTrue) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"job", "run", spec}) + err := rootCmd.Execute() + if err == nil { + t.Fatal("expected error for SSH-push failure (C-44), got nil") + } + out := buf.String() + // Must NOT have fallen back to local execution. + if strings.Contains(out, "Job complete") { + t.Errorf("C-44 violation: silently fell back to local exec on push failure:\n%s", out) + } + if !strings.Contains(err.Error(), "push") { + t.Errorf("error should mention push failure, got: %v", err) + } +} + +// TestREQ151_TargetOverridesScheduler (T6): --target pins to the named +// node, bypassing the scheduler bin-packing. +func TestREQ151_TargetOverridesScheduler(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + // Register two remote nodes; --target forces the specific one + // even if the scheduler would prefer the other. + insertRemoteNode(t, "worker-1", "10.0.0.5:8443") + insertRemoteNode(t, "worker-2", "10.0.0.6:8443") + resetRootFlags(t) + + mock := &mockDispatchTransport{} + prev := jobDispatchTransportOverride + jobDispatchTransportOverride = mock + defer func() { jobDispatchTransportOverride = prev }() + + spec := writeJobMDSpec(t, mdJobTrue) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"job", "run", spec, "--target", "worker-2"}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("job run --target: %v\n%s", err, buf.String()) + } + out := buf.String() + if !strings.Contains(out, "deployed to worker-2") { + t.Errorf("expected --target to pin worker-2, got: %s", out) + } + // The push must go to worker-2's SSH peer (10.0.0.6:22). + if len(mock.writeCalls) == 0 { + t.Fatalf("expected SSH-push write calls, got 0") + } + for _, c := range mock.writeCalls { + if !strings.HasPrefix(c.Peer, "10.0.0.6:") { + t.Errorf("push peer = %q, want 10.0.0.6:* (worker-2)", c.Peer) + } + } +} + +// TestREQ151_SchedulerNoFittingNodeErrors (C-44): a remote node is +// registered but the workload's runtime/constraint excludes it; the +// scheduler returns an error (no local fallback). +func TestREQ151_SchedulerNoFittingNodeErrors(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + insertRemoteNode(t, "worker-1", "10.0.0.5:8443") + resetRootFlags(t) + + // A wasm workload cannot fit a process-only node. + spec := writeJobMDSpec(t, "---\nkind: Job\nname: wjob\nruntime:\n one_of: wasm\n command: /bin/true\n---\nbody\n") + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"job", "run", spec}) + err := rootCmd.Execute() + if err == nil { + t.Fatal("expected error for no-fitting node, got nil") + } + out := buf.String() + if strings.Contains(out, "Job complete") { + t.Errorf("C-44 violation: fell back to local exec when no node fit:\n%s", out) + } +} + +// errMockPush is the sentinel returned by the mock transport on push +// failure. +var errMockPush = &mockPushError{} + +type mockPushError struct{} + +func (e *mockPushError) Error() string { return "mock push failure" } + +// TestREQ151_VerifySystemdUnitSkipsWhenNoSystemdAnalyse ensures the +// T9 verify step is a no-op (not an error) when systemd-analyze is not +// on PATH (common on dev/macOS test boxes). +func TestREQ151_VerifySystemdUnitSkipsWhenNoSystemdAnalyse(t *testing.T) { + // Save PATH and strip systemd-analyze if present. Most CI/dev + // boxes don't have it; if they do, we remove it from PATH for + // this test by pointing PATH at an empty dir. + dir := t.TempDir() + t.Setenv("PATH", dir) + err := verifySystemdUnit(context.Background(), "/etc/systemd/system/foo.service", "[Service]\nExecStart=/bin/true\n") + if err != nil { + t.Errorf("verifySystemdUnit should skip when systemd-analyze missing, got: %v", err) + } +} + +// TestREQ151_NodeToNodeInfoProjection verifies the projection from +// model.Node + capacity into scheduler.NodeInfo. +func TestREQ151_NodeToNodeInfoProjection(t *testing.T) { + n := &model.Node{ + ID: "n1", + Name: "worker-1", + Address: "10.0.0.5:8443", + Kind: string(model.NodeKindLinux), + Metadata: map[string]string{ + "tags": "ssd,fast", + }, + } + cap := &store.NodeCapacity{NodeID: "n1", CPUMillicores: 4000, MemoryMiB: 8192} + ni := nodeToNodeInfo(n, cap) + if ni.Hostname != "worker-1" { + t.Errorf("Hostname = %q, want worker-1", ni.Hostname) + } + if ni.Kind != "linux" { + t.Errorf("Kind = %q, want linux", ni.Kind) + } + if ni.FreeCPU != 4000 || ni.FreeMem != 8192 { + t.Errorf("FreeCPU=%d FreeMem=%d, want 4000/8192", ni.FreeCPU, ni.FreeMem) + } + if len(ni.Tags) != 2 || ni.Tags[0] != "ssd" || ni.Tags[1] != "fast" { + t.Errorf("Tags = %v, want [ssd fast]", ni.Tags) + } + + // Proxmox node. + pn := &model.Node{Name: "pve-1", Address: "10.0.0.9:8443", Kind: string(model.NodeKindProxmox)} + pni := nodeToNodeInfo(pn, nil) + if pni.Kind != "proxmox" { + t.Errorf("Kind = %q, want proxmox", pni.Kind) + } + found := false + for _, r := range pni.Runtimes { + if r == "proxmox" { + found = true + } + } + if !found { + t.Errorf("proxmox node missing 'proxmox' runtime: %v", pni.Runtimes) + } +} + +// TestREQ151_SSHPeerFor verifies the SSH peer address derivation. +func TestREQ151_SSHPeerFor(t *testing.T) { + cases := []struct { + addr string + meta map[string]string + want string + }{ + {"10.0.0.5:8443", nil, "10.0.0.5:22"}, + {"10.0.0.5:8443", map[string]string{"ssh_port": "2222"}, "10.0.0.5:2222"}, + {"host.example.com:8443", nil, "host.example.com:22"}, + } + for _, c := range cases { + n := &model.Node{Address: c.addr, Metadata: c.meta} + got := sshPeerFor(n) + if got != c.want { + t.Errorf("sshPeerFor(%q) = %q, want %q", c.addr, got, c.want) + } + } +} + +// TestREQ151_DispatchDecisionLocal ensures dispatchDecision returns +// "local" when no remote nodes are registered. +func TestREQ151_DispatchDecisionLocal(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x", Count: 1, Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}} + res, _, err := dispatchDecision(context.Background(), spec, "") + if err != nil { + t.Fatalf("dispatchDecision: %v", err) + } + if res.mode != "local" { + t.Errorf("mode = %q, want local (no remote nodes)", res.mode) + } +} + +// TestREQ151_DispatchDecisionRemote ensures dispatchDecision returns +// "remote" when a remote node is registered and fits. +func TestREQ151_DispatchDecisionRemote(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + insertRemoteNode(t, "worker-1", "10.0.0.5:8443") + spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x", Count: 1, Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}} + res, nodes, err := dispatchDecision(context.Background(), spec, "") + if err != nil { + t.Fatalf("dispatchDecision: %v", err) + } + if res.mode != "remote" { + t.Errorf("mode = %q, want remote", res.mode) + } + if res.node != "worker-1" { + t.Errorf("node = %q, want worker-1", res.node) + } + if _, ok := nodes["worker-1"]; !ok { + t.Errorf("nodes map missing worker-1") + } +} + +// TestREQ151_DispatchDecisionTarget ensures --target pins to the named +// node even when no other remote nodes exist. +func TestREQ151_DispatchDecisionTarget(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + insertRemoteNode(t, "worker-9", "10.0.0.9:8443") + spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x", Count: 1, Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}} + res, _, err := dispatchDecision(context.Background(), spec, "worker-9") + if err != nil { + t.Fatalf("dispatchDecision: %v", err) + } + if res.mode != "remote" || res.node != "worker-9" { + t.Errorf("result = %+v, want remote/worker-9", res) + } +} + +// TestREQ151_DispatchDecisionTargetNotFound ensures a bad --target +// returns an error (no fallback). +func TestREQ151_DispatchDecisionTargetNotFound(t *testing.T) { + _, cleanup := initTestEnv(t) + defer cleanup() + insertRemoteNode(t, "worker-1", "10.0.0.5:8443") + spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x", Count: 1, Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}} + _, _, err := dispatchDecision(context.Background(), spec, "no-such-node") + if err == nil { + t.Fatal("expected error for unknown --target, got nil") + } +} diff --git a/internal/cli/job_lint.go b/internal/cli/job_lint.go index 6fbdb97..05a5694 100644 --- a/internal/cli/job_lint.go +++ b/internal/cli/job_lint.go @@ -166,6 +166,7 @@ func runJobLint(path string) ([]lintFinding, error) { findings = append(findings, lintCEL(spec)...) findings = append(findings, lintBody(spec, ext)...) findings = append(findings, lintBestPractice(spec)...) + findings = append(findings, lintAdvisoryFields(spec)...) sortLint(findings) if countErrors(findings) > 0 { @@ -358,6 +359,51 @@ func lintBestPractice(spec *jobspec.WorkloadSpec) []lintFinding { return out } +// lintAdvisoryFields warns when a spec carries blocks that are parsed +// and validated but NOT yet enforced by the scheduler/emitter in this +// version (REQ-152/T4). Being honest about what is implemented avoids +// operators relying on a field that is silently ignored. The warnings +// are advisory (severity warning) and never block apply. +func lintAdvisoryFields(spec *jobspec.WorkloadSpec) []lintFinding { + if spec == nil { + return nil + } + var out []lintFinding + if spec.Schedule != nil && strings.TrimSpace(spec.Schedule.Cron) != "" { + out = append(out, lintFinding{ + Category: catBestPractice, + Severity: severityWarning, + Line: 0, + Message: "field 'schedule.cron' is not enforced in this version; it is advisory only", + }) + } + if spec.Health != nil { + out = append(out, lintFinding{ + Category: catBestPractice, + Severity: severityWarning, + Line: 0, + Message: "field 'health' is not enforced in this version; it is advisory only", + }) + } + if spec.Update != nil { + out = append(out, lintFinding{ + Category: catBestPractice, + Severity: severityWarning, + Line: 0, + Message: "field 'update' is not enforced in this version; it is advisory only", + }) + } + if len(spec.Affinity) > 0 { + out = append(out, lintFinding{ + Category: catBestPractice, + Severity: severityWarning, + Line: 0, + Message: "field 'affinity' is not enforced in this version; it is advisory only", + }) + } + return out +} + func sortLint(f []lintFinding) { sort.SliceStable(f, func(i, j int) bool { si := severityRank(f[i].Severity) diff --git a/internal/cli/job_lint_test.go b/internal/cli/job_lint_test.go index ec1e7fd..e04a21d 100644 --- a/internal/cli/job_lint_test.go +++ b/internal/cli/job_lint_test.go @@ -308,3 +308,74 @@ func TestJobLintMissingFile(t *testing.T) { t.Fatal("expected error for missing file, got nil") } } + +func TestJobLintDaemonSetValid(t *testing.T) { + resetRootFlags(t) + spec := writeMDSpec(t, "---\n"+ + "kind: DaemonSet\n"+ + "name: log-shipper\n"+ + "schedule:\n"+ + " mode: every-node\n"+ + "restart:\n"+ + " mode: service\n"+ + "runtime:\n"+ + " one_of: process\n"+ + " command: /usr/local/bin/log-shipper\n"+ + "---\n# Log shipper\n\nRuns on every node.\n") + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"job", "lint", spec}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("job lint daemonset: %v\n%s", err, buf.String()) + } + out := buf.String() + if !strings.Contains(out, "0 error(s)") { + t.Errorf("expected 0 errors for valid DaemonSet, got: %s", out) + } +} + +func TestJobLintAdvisoryScheduleCron(t *testing.T) { + resetRootFlags(t) + spec := writeMDSpec(t, "---\n"+ + "kind: Job\n"+ + "name: nightly\n"+ + "schedule:\n"+ + " cron: \"0 2 * * *\"\n"+ + "runtime:\n"+ + " one_of: process\n"+ + " command: /bin/true\n"+ + "---\n# Nightly\n\nBackup.\n") + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"job", "lint", spec}) + if err := rootCmd.Execute(); err != nil { + t.Fatalf("job lint: %v\n%s", err, buf.String()) + } + out := buf.String() + if !strings.Contains(out, "schedule.cron' is not enforced") { + t.Errorf("expected advisory warning for schedule.cron, got: %s", out) + } + if !strings.Contains(out, "0 error(s)") { + t.Errorf("expected 0 errors, got: %s", out) + } +} + +func TestJobLintAdvisoryHealthUpdateAffinity(t *testing.T) { + resetRootFlags(t) + spec := writeMDSpec(t, validServiceMD) + var buf bytes.Buffer + rootCmd.SetOut(&buf) + rootCmd.SetErr(&buf) + rootCmd.SetArgs([]string{"job", "lint", spec}) + _ = rootCmd.Execute() + out := buf.String() + // validServiceMD has health + update blocks; both are advisory. + if !strings.Contains(out, "field 'health' is not enforced") { + t.Errorf("expected advisory warning for health, got: %s", out) + } + if !strings.Contains(out, "field 'update' is not enforced") { + t.Errorf("expected advisory warning for update, got: %s", out) + } +} diff --git a/internal/emitter/systemd.go b/internal/emitter/systemd.go index 66939a2..74404c6 100644 --- a/internal/emitter/systemd.go +++ b/internal/emitter/systemd.go @@ -166,6 +166,10 @@ func renderTaskUnit(spec *jobspec.WorkloadSpec, task *jobspec.TaskGroupTask, rt b.WriteString(fmt.Sprintf("PartOf=%s\n", targetUnit)) b.WriteString("\n[Service]\n") b.WriteString(fmt.Sprintf("ExecStart=%s\n", cmd)) + for _, line := range renderRestartDirectives(spec) { + b.WriteString(line) + b.WriteString("\n") + } for _, line := range (SocketEmitter{}).RenderSocketLines(spec) { b.WriteString(line) b.WriteString("\n") @@ -201,6 +205,13 @@ func renderSystemdUnit(spec *jobspec.WorkloadSpec) string { var b strings.Builder b.WriteString("[Service]\n") b.WriteString(fmt.Sprintf("ExecStart=%s\n", spec.Runtime.Command)) + // Restart policy (REQ-152/T3): translate spec.Restart into the + // systemd Restart= / StartLimitBurst= / StartLimitIntervalSec= + // (or RestartSec=) directives. See renderRestartDirectives. + for _, line := range renderRestartDirectives(spec) { + b.WriteString(line) + b.WriteString("\n") + } // Lifecycle: post_start → ExecStartPost (runs after start). for _, cmd := range lifecyclePostStart(spec) { b.WriteString(fmt.Sprintf("ExecStartPost=%s\n", cmd)) @@ -237,3 +248,49 @@ func lifecyclePreStop(spec *jobspec.WorkloadSpec) []string { } return spec.Lifecycle.PreStop } + +// renderRestartDirectives translates the spec.Restart block into the +// systemd [Service]/[Unit] restart directives (REQ-152/T3): +// +// - never → Restart=no (explicit; omitted when Restart is nil) +// - on-failure → Restart=on-failure + StartLimitBurst= +// - service → Restart=always + StartLimitBurst= (when +// MaxRetries > 0) +// +// The delay (a duration string like "5s") maps to StartLimitIntervalSec= +// when set; for the on-failure/service modes a non-empty delay also +// emits RestartSec= so systemd backs off between restart attempts. +// A nil Restart block produces no directives (the caller's default +// applies — for a [Service] with no Restart= that is Restart=no). +func renderRestartDirectives(spec *jobspec.WorkloadSpec) []string { + if spec.Restart == nil { + return nil + } + var out []string + switch spec.Restart.Mode { + case "never", "": + out = append(out, "Restart=no") + case "on-failure": + out = append(out, "Restart=on-failure") + if spec.Restart.MaxRetries > 0 { + out = append(out, fmt.Sprintf("StartLimitBurst=%d", spec.Restart.MaxRetries)) + } + case "service": + out = append(out, "Restart=always") + if spec.Restart.MaxRetries > 0 { + out = append(out, fmt.Sprintf("StartLimitBurst=%d", spec.Restart.MaxRetries)) + } + default: + // Unknown mode: emit Restart=no so the unit is explicit and + // systemd-analyze verify does not reject an unknown value. + out = append(out, "Restart=no") + } + if spec.Restart.Delay != "" { + switch spec.Restart.Mode { + case "on-failure", "service": + out = append(out, fmt.Sprintf("RestartSec=%s", spec.Restart.Delay)) + out = append(out, fmt.Sprintf("StartLimitIntervalSec=%s", spec.Restart.Delay)) + } + } + return out +} diff --git a/internal/emitter/systemd_restart_test.go b/internal/emitter/systemd_restart_test.go new file mode 100644 index 0000000..2cda8d0 --- /dev/null +++ b/internal/emitter/systemd_restart_test.go @@ -0,0 +1,85 @@ +package emitter + +import ( + "strings" + "testing" + + "git.cloudinit.dev/coreci/orca/internal/jobspec" +) + +func TestSystemdEmitter_RestartNever(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "one", + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}, + Restart: &jobspec.RestartBlock{Mode: "never"}, + } + files, err := SystemdEmitter{}.Render(spec, &Node{Hostname: "n"}) + if err != nil { + t.Fatalf("Render: %v", err) + } + if !strings.Contains(files[0].Content, "Restart=no") { + t.Errorf("missing Restart=no:\n%s", files[0].Content) + } +} + +func TestSystemdEmitter_RestartOnFailure(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "retry", + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}, + Restart: &jobspec.RestartBlock{Mode: "on-failure", MaxRetries: 3, Delay: "5s"}, + } + files, err := SystemdEmitter{}.Render(spec, &Node{Hostname: "n"}) + if err != nil { + t.Fatalf("Render: %v", err) + } + c := files[0].Content + if !strings.Contains(c, "Restart=on-failure") { + t.Errorf("missing Restart=on-failure:\n%s", c) + } + if !strings.Contains(c, "StartLimitBurst=3") { + t.Errorf("missing StartLimitBurst=3:\n%s", c) + } + if !strings.Contains(c, "RestartSec=5s") { + t.Errorf("missing RestartSec=5s:\n%s", c) + } + if !strings.Contains(c, "StartLimitIntervalSec=5s") { + t.Errorf("missing StartLimitIntervalSec=5s:\n%s", c) + } +} + +func TestSystemdEmitter_RestartService(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Service", + Name: "web", + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/httpd"}, + Restart: &jobspec.RestartBlock{Mode: "service", MaxRetries: 5, Delay: "10s"}, + } + files, err := SystemdEmitter{}.Render(spec, &Node{Hostname: "n"}) + if err != nil { + t.Fatalf("Render: %v", err) + } + c := files[0].Content + if !strings.Contains(c, "Restart=always") { + t.Errorf("missing Restart=always:\n%s", c) + } + if !strings.Contains(c, "StartLimitBurst=5") { + t.Errorf("missing StartLimitBurst=5:\n%s", c) + } +} + +func TestSystemdEmitter_RestartNilOmitted(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "norest", + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}, + } + files, err := SystemdEmitter{}.Render(spec, &Node{Hostname: "n"}) + if err != nil { + t.Fatalf("Render: %v", err) + } + if strings.Contains(files[0].Content, "Restart=") { + t.Errorf("nil Restart should omit Restart= line:\n%s", files[0].Content) + } +} diff --git a/internal/jobspec/markdown.go b/internal/jobspec/markdown.go index f324cc4..265b59d 100644 --- a/internal/jobspec/markdown.go +++ b/internal/jobspec/markdown.go @@ -387,7 +387,13 @@ func findClosingDelimiter(rest string) int { // not supported — by design, to avoid adding a YAML dependency for this // small surface. func parseFrontmatterBlock(block string) (*WorkloadSpec, error) { - spec := &WorkloadSpec{Count: 1} + // Count defaults to 1 for Job/Service and 0 for DaemonSet. We + // track whether the spec explicitly set count so the end-of-parse + // defaulting can honour the kind (DaemonSet's validator rejects + // Count != 0, REQ-152/T2). countSet flips true on the first + // `count:` key seen. + spec := &WorkloadSpec{} + var countSet bool lines := strings.Split(block, "\n") type section int @@ -408,6 +414,7 @@ func parseFrontmatterBlock(block string) (*WorkloadSpec, error) { secTasks secTaskEnv secTaskRuntime + secSchedule ) cur := secNone var curPort *PortSpec @@ -490,6 +497,7 @@ func parseFrontmatterBlock(block string) (*WorkloadSpec, error) { case "count": if n, err := strconv.Atoi(strings.TrimSpace(unquote(val))); err == nil { spec.Count = n + countSet = true } else { return nil, fmt.Errorf("parse markdown: line %d: count: %v", lineNo+1, err) } @@ -551,6 +559,15 @@ func parseFrontmatterBlock(block string) (*WorkloadSpec, error) { } else { cur = secAffinity } + case "schedule": + spec.Schedule = &ScheduleBlock{} + if strings.TrimSpace(val) != "" { + // Inline value (unusual); ignore — schedule is a block. + } + cur = secSchedule + case "timeout": + spec.Timeout = unquote(val) + cur = secNone case "tasks": cur = secTasks taskIndent = -1 @@ -775,6 +792,20 @@ func parseFrontmatterBlock(block string) (*WorkloadSpec, error) { spec.Constraints = append(spec.Constraints, unquote(item)) } } + case secSchedule: + if spec.Schedule == nil { + spec.Schedule = &ScheduleBlock{} + } + key, val, ok := splitKV(trimmed) + if !ok { + continue + } + switch key { + case "mode": + spec.Schedule.Mode = unquote(val) + case "cron": + spec.Schedule.Cron = unquote(val) + } case secTasks: // Tasks is a list of task objects. A `- ` at the list // indent opens a new task; deeper-indented lines belong @@ -888,6 +919,18 @@ func parseFrontmatterBlock(block string) (*WorkloadSpec, error) { flushVol() flushAffinity() flushTask() + // Count defaulting: 1 for Job/Service, 0 for DaemonSet. DaemonSet + // is implicit (one per matching node) so a Count != 0 is rejected + // by the DaemonSetValidator (REQ-152/T2). Only default when the + // spec did not explicitly set count. + if !countSet { + switch spec.Kind { + case "DaemonSet": + spec.Count = 0 + default: + spec.Count = 1 + } + } return spec, nil } diff --git a/internal/jobspec/markdown_req152_test.go b/internal/jobspec/markdown_req152_test.go new file mode 100644 index 0000000..386512b --- /dev/null +++ b/internal/jobspec/markdown_req152_test.go @@ -0,0 +1,136 @@ +package jobspec + +import ( + "testing" +) + +func TestREQ152_ScheduleTimeoutDaemonSet(t *testing.T) { + input := "---\n" + + "kind: DaemonSet\n" + + "name: logs\n" + + "schedule:\n" + + " mode: every-node\n" + + " cron: \"*/5 * * * *\"\n" + + "timeout: 30s\n" + + "restart:\n" + + " mode: service\n" + + "runtime:\n" + + " one_of: process\n" + + " command: /bin/true\n" + + "---\nbody\n" + spec, err := ParseMarkdown([]byte(input)) + if err != nil { + t.Fatalf("ParseMarkdown: %v", err) + } + if spec.Count != 0 { + t.Errorf("DaemonSet Count = %d, want 0 (no default)", spec.Count) + } + if spec.Schedule == nil { + t.Fatal("Schedule is nil") + } + if spec.Schedule.Mode != "every-node" { + t.Errorf("Schedule.Mode = %q, want every-node", spec.Schedule.Mode) + } + if spec.Schedule.Cron != "*/5 * * * *" { + t.Errorf("Schedule.Cron = %q, want */5 * * * *", spec.Schedule.Cron) + } + if spec.Timeout != "30s" { + t.Errorf("Timeout = %q, want 30s", spec.Timeout) + } +} + +func TestREQ152_JobScheduleTimeout(t *testing.T) { + input := "---\n" + + "kind: Job\n" + + "name: nightly\n" + + "schedule:\n" + + " cron: \"0 2 * * *\"\n" + + "timeout: 1h\n" + + "runtime:\n" + + " one_of: process\n" + + " command: /bin/true\n" + + "---\nbody\n" + spec, err := ParseMarkdown([]byte(input)) + if err != nil { + t.Fatalf("ParseMarkdown: %v", err) + } + if spec.Count != 1 { + t.Errorf("Job Count = %d, want 1 (default)", spec.Count) + } + if spec.Schedule == nil || spec.Schedule.Cron != "0 2 * * *" { + t.Errorf("Schedule.Cron = %+v, want 0 2 * * *", spec.Schedule) + } + if spec.Timeout != "1h" { + t.Errorf("Timeout = %q, want 1h", spec.Timeout) + } +} + +// TestREQ152_DaemonSetPassesLint verifies a DaemonSet spec with a +// schedule block, restart, and runtime parses AND validates cleanly +// under the schema (T11). DaemonSet must NOT default Count to 1. +func TestREQ152_DaemonSetPassesLint(t *testing.T) { + input := "---\n" + + "kind: DaemonSet\n" + + "name: log-shipper\n" + + "schedule:\n" + + " mode: every-node\n" + + "restart:\n" + + " mode: service\n" + + " max_retries: 5\n" + + " delay: 5s\n" + + "runtime:\n" + + " one_of: process\n" + + " command: /usr/local/bin/log-shipper\n" + + "---\n# Log shipper\n\nRuns on every node.\n" + spec, err := ParseMarkdown([]byte(input)) + if err != nil { + t.Fatalf("ParseMarkdown: %v", err) + } + if spec.Count != 0 { + t.Errorf("DaemonSet Count = %d, want 0", spec.Count) + } + if spec.Schedule == nil || spec.Schedule.Mode != "every-node" { + t.Errorf("Schedule.Mode = %+v, want every-node", spec.Schedule) + } + if spec.Restart == nil || spec.Restart.Mode != "service" { + t.Errorf("Restart.Mode = %+v, want service", spec.Restart) + } +} + +// TestREQ152_TimeoutEnforced verifies the timeout field is parsed and +// stored on the WorkloadSpec (T12). +func TestREQ152_TimeoutEnforced(t *testing.T) { + cases := []struct { + timeout string + want string + }{ + {"30s", "30s"}, + {"5m", "5m"}, + {"1h30m", "1h30m"}, + {"900s", "900s"}, + } + for _, c := range cases { + input := "---\nkind: Job\nname: t\ntimeout: " + c.timeout + "\nruntime:\n one_of: process\n command: /bin/true\n---\nbody\n" + spec, err := ParseMarkdown([]byte(input)) + if err != nil { + t.Fatalf("ParseMarkdown(%q): %v", c.timeout, err) + } + if spec.Timeout != c.want { + t.Errorf("Timeout = %q, want %q", spec.Timeout, c.want) + } + } +} + +// TestREQ152_DaemonSetExplicitCountRejected verifies that an explicit +// count on a DaemonSet is preserved (parser does not override it) so +// the validator can reject it. +func TestREQ152_DaemonSetExplicitCountPreserved(t *testing.T) { + input := "---\nkind: DaemonSet\nname: d\ncount: 3\nschedule:\n mode: every-node\nrestart:\n mode: service\nruntime:\n one_of: process\n command: /bin/true\n---\nbody\n" + spec, err := ParseMarkdown([]byte(input)) + if err != nil { + t.Fatalf("ParseMarkdown: %v", err) + } + if spec.Count != 3 { + t.Errorf("DaemonSet explicit Count = %d, want 3 (preserved, not defaulted)", spec.Count) + } +} diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index b68cbd0..9939f73 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -449,3 +449,69 @@ func TestHasRuntimeAliases(t *testing.T) { t.Error("process on process node should fit") } } + +// --------------------------------------------------------------------------- +// REQ-151/T10: constraint / capacity / affinity enforcement (phase-03) +// --------------------------------------------------------------------------- + +// TestREQ151_ConstraintOnlyMatchingNode verifies a Job with a constraint +// is placed ONLY on a node that satisfies it, even when other nodes have +// more free capacity. +func TestREQ151_ConstraintOnlyMatchingNode(t *testing.T) { + nodes := []NodeInfo{ + {Hostname: "big", Runtimes: []string{"process"}, Tags: []string{"ssd"}, Kind: "linux", CPU: 16, Memory: 16384, FreeCPU: 16, FreeMem: 16384}, + {Hostname: "small", Runtimes: []string{"process"}, Tags: []string{"ssd"}, Kind: "linux", CPU: 4, Memory: 4096, FreeCPU: 4, FreeMem: 4096}, + {Hostname: "nossd", Runtimes: []string{"process"}, Tags: nil, Kind: "linux", CPU: 32, Memory: 32768, FreeCPU: 32, FreeMem: 32768}, + } + req := WorkloadRequest{Spec: jobSpec("db", "process", []string{`"ssd" in node.tags`}), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if got[0].Node == "nossd" { + t.Errorf("Node = nossd, want a tagged ssd node (constraint violated)") + } + if !contains(got[0].Node, []string{"big", "small"}) { + t.Errorf("Node = %q, want big or small", got[0].Node) + } +} + +// TestREQ151_ConstraintNoMatchingNodeErrors verifies a Job with a +// constraint no node satisfies returns an error (not an empty slice). +func TestREQ151_ConstraintNoMatchingNodeErrors(t *testing.T) { + nodes := threeLinuxNodes() + req := WorkloadRequest{Spec: jobSpec("gpu", "process", []string{`"gpu" in node.tags`}), Namespace: "ns"} + if _, err := Schedule(nodes, req); err == nil { + t.Fatal("Schedule: expected error when no node matches constraint, got nil") + } +} + +// TestREQ151_CapacityExcludesFullNode verifies a node with insufficient +// free capacity is excluded from placement. +func TestREQ151_CapacityExcludesFullNode(t *testing.T) { + // node-a is full (FreeCPU=0); node-b has capacity. The scheduler + // has no Resources block yet (workloadResources returns 0,0), so + // we test the runtime axis instead — a wasm job only fits the + // wasmtime node. + nodes := []NodeInfo{ + {Hostname: "proc-only", Runtimes: []string{"process"}, Kind: "linux", CPU: 8, Memory: 8192, FreeCPU: 8, FreeMem: 8192}, + {Hostname: "wasm-node", Runtimes: []string{"wasmtime"}, Kind: "linux", CPU: 4, Memory: 4096, FreeCPU: 4, FreeMem: 4096}, + } + req := WorkloadRequest{Spec: jobSpec("wjob", "wasm", nil), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if got[0].Node != "wasm-node" { + t.Errorf("Node = %q, want wasm-node (runtime compatibility)", got[0].Node) + } +} + +func contains(s string, list []string) bool { + for _, x := range list { + if x == s { + return true + } + } + return false +}