Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| cf3d98eb2b |
+74
-24
@@ -60,16 +60,21 @@ var jobRunCmd = &cobra.Command{
|
|||||||
ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Minute)
|
ctx, cancel := context.WithTimeout(cmd.Context(), 5*time.Minute)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
exec, closer, err := jobExecutor()
|
// v0.13 phase-03 scheduler wiring (REQ-151, C-44): decide
|
||||||
if err != nil {
|
// whether to run locally (dev mode / no remote nodes) or
|
||||||
return err
|
// remotely (scheduler picks a peer, render systemd, SSH-push).
|
||||||
}
|
// The deprecated mTLS Dispatcher path (--idempotency-key) is
|
||||||
defer closer()
|
// retained only for the dual-write window; the new remote path
|
||||||
|
// uses the CLI-side scheduler + sshpush.
|
||||||
// If --target or --idempotency-key is set, route through the
|
if runIDKey != "" {
|
||||||
// dispatcher (which may land the job locally or on a peer
|
// Legacy --idempotency-key dispatch path (deprecated mTLS
|
||||||
// based on capacity).
|
// Dispatcher). Retained for backward compat; routes through
|
||||||
if runTarget != "" || runIDKey != "" {
|
// 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()
|
db, dbCloser, err := openDB()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -95,25 +100,70 @@ var jobRunCmd = &cobra.Command{
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
job := &model.Job{
|
res, nodesByHost, err := dispatchDecision(ctx, spec, runTarget)
|
||||||
ID: uuid.NewString(),
|
if err != nil {
|
||||||
Name: spec.Name,
|
logDispatch(nil, err)
|
||||||
Spec: args[0],
|
|
||||||
Status: model.JobStatusPending,
|
|
||||||
}
|
|
||||||
if err := exec.Run(ctx, job, workloadToTaskSpecs(spec)); err != nil {
|
|
||||||
if jsonOutput {
|
if jsonOutput {
|
||||||
_ = printJSON(map[string]any{"id": job.ID, "status": "failed", "error": err.Error()})
|
_ = printJSON(map[string]any{"status": "failed", "error": err.Error()})
|
||||||
return err
|
|
||||||
}
|
}
|
||||||
fmt.Fprintf(cmd.ErrOrStderr(), "✗ Job %s failed: %v\n", job.ID, err)
|
|
||||||
return 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 fmt.Errorf("job run: unknown dispatch mode %q", res.mode)
|
||||||
return nil
|
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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...)
|
||||||
|
}
|
||||||
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -166,6 +166,7 @@ func runJobLint(path string) ([]lintFinding, error) {
|
|||||||
findings = append(findings, lintCEL(spec)...)
|
findings = append(findings, lintCEL(spec)...)
|
||||||
findings = append(findings, lintBody(spec, ext)...)
|
findings = append(findings, lintBody(spec, ext)...)
|
||||||
findings = append(findings, lintBestPractice(spec)...)
|
findings = append(findings, lintBestPractice(spec)...)
|
||||||
|
findings = append(findings, lintAdvisoryFields(spec)...)
|
||||||
|
|
||||||
sortLint(findings)
|
sortLint(findings)
|
||||||
if countErrors(findings) > 0 {
|
if countErrors(findings) > 0 {
|
||||||
@@ -358,6 +359,51 @@ func lintBestPractice(spec *jobspec.WorkloadSpec) []lintFinding {
|
|||||||
return out
|
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) {
|
func sortLint(f []lintFinding) {
|
||||||
sort.SliceStable(f, func(i, j int) bool {
|
sort.SliceStable(f, func(i, j int) bool {
|
||||||
si := severityRank(f[i].Severity)
|
si := severityRank(f[i].Severity)
|
||||||
|
|||||||
@@ -308,3 +308,74 @@ func TestJobLintMissingFile(t *testing.T) {
|
|||||||
t.Fatal("expected error for missing file, got nil")
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -166,6 +166,10 @@ func renderTaskUnit(spec *jobspec.WorkloadSpec, task *jobspec.TaskGroupTask, rt
|
|||||||
b.WriteString(fmt.Sprintf("PartOf=%s\n", targetUnit))
|
b.WriteString(fmt.Sprintf("PartOf=%s\n", targetUnit))
|
||||||
b.WriteString("\n[Service]\n")
|
b.WriteString("\n[Service]\n")
|
||||||
b.WriteString(fmt.Sprintf("ExecStart=%s\n", cmd))
|
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) {
|
for _, line := range (SocketEmitter{}).RenderSocketLines(spec) {
|
||||||
b.WriteString(line)
|
b.WriteString(line)
|
||||||
b.WriteString("\n")
|
b.WriteString("\n")
|
||||||
@@ -201,6 +205,13 @@ func renderSystemdUnit(spec *jobspec.WorkloadSpec) string {
|
|||||||
var b strings.Builder
|
var b strings.Builder
|
||||||
b.WriteString("[Service]\n")
|
b.WriteString("[Service]\n")
|
||||||
b.WriteString(fmt.Sprintf("ExecStart=%s\n", spec.Runtime.Command))
|
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).
|
// Lifecycle: post_start → ExecStartPost (runs after start).
|
||||||
for _, cmd := range lifecyclePostStart(spec) {
|
for _, cmd := range lifecyclePostStart(spec) {
|
||||||
b.WriteString(fmt.Sprintf("ExecStartPost=%s\n", cmd))
|
b.WriteString(fmt.Sprintf("ExecStartPost=%s\n", cmd))
|
||||||
@@ -237,3 +248,49 @@ func lifecyclePreStop(spec *jobspec.WorkloadSpec) []string {
|
|||||||
}
|
}
|
||||||
return spec.Lifecycle.PreStop
|
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=<MaxRetries>
|
||||||
|
// - service → Restart=always + StartLimitBurst=<MaxRetries> (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=<delay> 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
|
||||||
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -387,7 +387,13 @@ func findClosingDelimiter(rest string) int {
|
|||||||
// not supported — by design, to avoid adding a YAML dependency for this
|
// not supported — by design, to avoid adding a YAML dependency for this
|
||||||
// small surface.
|
// small surface.
|
||||||
func parseFrontmatterBlock(block string) (*WorkloadSpec, error) {
|
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")
|
lines := strings.Split(block, "\n")
|
||||||
|
|
||||||
type section int
|
type section int
|
||||||
@@ -408,6 +414,7 @@ func parseFrontmatterBlock(block string) (*WorkloadSpec, error) {
|
|||||||
secTasks
|
secTasks
|
||||||
secTaskEnv
|
secTaskEnv
|
||||||
secTaskRuntime
|
secTaskRuntime
|
||||||
|
secSchedule
|
||||||
)
|
)
|
||||||
cur := secNone
|
cur := secNone
|
||||||
var curPort *PortSpec
|
var curPort *PortSpec
|
||||||
@@ -490,6 +497,7 @@ func parseFrontmatterBlock(block string) (*WorkloadSpec, error) {
|
|||||||
case "count":
|
case "count":
|
||||||
if n, err := strconv.Atoi(strings.TrimSpace(unquote(val))); err == nil {
|
if n, err := strconv.Atoi(strings.TrimSpace(unquote(val))); err == nil {
|
||||||
spec.Count = n
|
spec.Count = n
|
||||||
|
countSet = true
|
||||||
} else {
|
} else {
|
||||||
return nil, fmt.Errorf("parse markdown: line %d: count: %v", lineNo+1, err)
|
return nil, fmt.Errorf("parse markdown: line %d: count: %v", lineNo+1, err)
|
||||||
}
|
}
|
||||||
@@ -551,6 +559,15 @@ func parseFrontmatterBlock(block string) (*WorkloadSpec, error) {
|
|||||||
} else {
|
} else {
|
||||||
cur = secAffinity
|
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":
|
case "tasks":
|
||||||
cur = secTasks
|
cur = secTasks
|
||||||
taskIndent = -1
|
taskIndent = -1
|
||||||
@@ -775,6 +792,20 @@ func parseFrontmatterBlock(block string) (*WorkloadSpec, error) {
|
|||||||
spec.Constraints = append(spec.Constraints, unquote(item))
|
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:
|
case secTasks:
|
||||||
// Tasks is a list of task objects. A `- ` at the list
|
// Tasks is a list of task objects. A `- ` at the list
|
||||||
// indent opens a new task; deeper-indented lines belong
|
// indent opens a new task; deeper-indented lines belong
|
||||||
@@ -888,6 +919,18 @@ func parseFrontmatterBlock(block string) (*WorkloadSpec, error) {
|
|||||||
flushVol()
|
flushVol()
|
||||||
flushAffinity()
|
flushAffinity()
|
||||||
flushTask()
|
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
|
return spec, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -449,3 +449,69 @@ func TestHasRuntimeAliases(t *testing.T) {
|
|||||||
t.Error("process on process node should fit")
|
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
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user