Files
orca/internal/spec/schema/schema.go
T
Jon Chery c3819dde12 feat(P06): task groups — multi-process services, multiple systemd units per alloc
P06 — Task groups (PRD §9.1: multiple systemd units per alloc).

Parser (internal/jobspec/markdown.go):
- TaskGroupTask type (Name, Runtime, Env, Command). Tasks []TaskGroupTask on
  WorkloadSpec. Parses tasks: frontmatter block (array of task objects).
  Tasks without their own runtime inherit the top-level Runtime as default.
  Backward compat: no tasks -> single-process (existing runtime block).

Systemd emitter (internal/emitter/systemd.go):
- Task group renders one systemd unit per task (orca-v1-alloc-<id>-<task>
  .service) plus a grouping target unit (orca-v1-alloc-<id>.target). Each
  per-task unit carries PartOf=<target> and WantedBy=multi-user.target.
  Single-process case unchanged (backward compat).

Schema (internal/spec/schema/schema.go):
- TaskGroup validation: unique task names, resolvable command (own or
  inherited). JobValidator/ServiceValidator/DaemonSetValidator all accept
  task groups.

Tests: 9 task-group tests in schema_test.go, lifecycle + target-unit tests
in systemd_test.go, parser tests in markdown_test.go. 22 packages pass.

Fix: 3 Service task-group test fixtures missing Count:1 (ServiceValidator
requires count>=1; a task-group Service still has >=1 replica).

---ci---
project: orca
phase: P06
milestone: v0.9
status: execute
---/ci---
2026-08-05 18:20:05 +00:00

255 lines
9.4 KiB
Go

// Package schema provides kind-specific validators for the unified
// *jobspec.WorkloadSpec introduced in P0b (REQ-064). Each workload kind
// (Job, Service, DaemonSet per R-012) has different required fields;
// this package exposes a Validator interface and a ValidatorFor
// dispatcher so the emitter layer (REQ-074) and the lint engine
// (REQ-084) can reject invalid specs before rendering.
//
// The validators operate purely on the *WorkloadSpec shape; they do no
// I/O. Required-field violations return a structured error listing
// every problem found (missing required fields, invalid combinations).
package schema
import (
"errors"
"fmt"
"net"
"strings"
"git.cloudinit.dev/coreci/orca/internal/jobspec"
)
// Validator validates a *jobspec.WorkloadSpec against a kind-specific
// schema. Implementations are pure (no I/O) and return a clear error
// listing every violation found.
type Validator interface {
Validate(spec *jobspec.WorkloadSpec) error
}
// JobValidator validates the Job workload kind (R-012).
//
// Rules:
// - no service block required (Job has no Traefik route by default D-175)
// - restart optional (defaults to never/on-failure when omitted)
// - schedule optional (cron string)
// - timeout optional
// - ports optional
// - count must be 1 (or unset → 1); count > 1 is an error for Job
// (use a Service for replicas)
// - no Traefik route (a ServiceBlock is rejected)
// - task group (spec.Tasks) optional; when present, each task must
// have a unique name and a resolvable command (P06).
type JobValidator struct{}
// ServiceValidator validates the Service workload kind (R-012).
//
// Rules:
// - ports required (at least one)
// - count ≥ 1
// - restart required (mode must be service)
// - update required (strategy must be rolling/canary/blue-green)
// - runtime required (unless a task group is present; each task
// can carry its own runtime — P06)
// - health block required (Traefik routing depends on health checks)
// - service block, if present, must have a valid bind (127.0.0.1
// opt-in per R-007; default is socket — empty bind is OK)
// - service block implied (Traefik route YES)
// - task group (spec.Tasks) optional; when present, each task must
// have a unique name and a resolvable command (P06).
type ServiceValidator struct{}
// DaemonSetValidator validates the DaemonSet workload kind (R-012).
//
// Rules:
// - schedule block with mode (every-node/matching/mandatory) required
// - no ports (no Traefik route by default D-175)
// - no count (implicit = nodes matching condition)
// - restart required
// - task group (spec.Tasks) optional; when present, each task must
// have a unique name and a resolvable command (P06).
type DaemonSetValidator struct{}
// ValidatorFor returns the Validator for the given workload kind, or an
// error for an unknown kind. kind must be one of Job, Service,
// DaemonSet (R-012).
func ValidatorFor(kind string) (Validator, error) {
switch kind {
case "Job":
return JobValidator{}, nil
case "Service":
return ServiceValidator{}, nil
case "DaemonSet":
return DaemonSetValidator{}, nil
default:
return nil, fmt.Errorf("schema: unknown kind %q (want one of Job, Service, DaemonSet)", kind)
}
}
// Validate validates a Job spec. See JobValidator for the rules.
func (JobValidator) Validate(spec *jobspec.WorkloadSpec) error {
if spec == nil {
return errors.New("schema/Job: spec is nil")
}
var errs []string
if strings.TrimSpace(spec.Name) == "" {
errs = append(errs, "name is required")
}
if spec.Count != 0 && spec.Count != 1 {
errs = append(errs, fmt.Sprintf("count must be 1 (or unset) for Job, got %d (use Service for replicas)", spec.Count))
}
if spec.Service != nil {
errs = append(errs, "service block (Traefik route) is not allowed for Job (D-175)")
}
errs = append(errs, validateTaskGroup(spec)...)
return composeErrors("schema/Job", errs)
}
// Validate validates a Service spec. See ServiceValidator for the rules.
func (ServiceValidator) Validate(spec *jobspec.WorkloadSpec) error {
if spec == nil {
return errors.New("schema/Service: spec is nil")
}
var errs []string
if strings.TrimSpace(spec.Name) == "" {
errs = append(errs, "name is required")
}
if len(spec.Ports) == 0 {
errs = append(errs, "ports required (at least one)")
}
if spec.Count < 1 {
errs = append(errs, fmt.Sprintf("count must be ≥ 1 for Service, got %d", spec.Count))
}
if spec.Restart == nil {
errs = append(errs, "restart block required for Service")
} else {
switch spec.Restart.Mode {
case "service", "on-failure", "never":
// Valid per R-012 (default for Service is "service",
// but the validator accepts the full enum; the
// Service-specific "must be service" rule is enforced
// below for the default case where mode is empty).
case "":
errs = append(errs, "restart mode required for Service (one of service, on-failure, never; default is service)")
default:
errs = append(errs, fmt.Sprintf("restart mode %q invalid (want one of service, on-failure, never)", spec.Restart.Mode))
}
}
if spec.Update == nil {
errs = append(errs, "update block required for Service")
} else {
switch spec.Update.Strategy {
case "rolling", "canary", "blue-green":
case "":
errs = append(errs, "update strategy required for Service (one of rolling, canary, blue-green)")
default:
errs = append(errs, fmt.Sprintf("update strategy %q invalid (want one of rolling, canary, blue-green)", spec.Update.Strategy))
}
}
if spec.Runtime == nil && len(spec.Tasks) == 0 {
errs = append(errs, "runtime block required for Service (or a task group with per-task runtimes)")
}
if spec.Health == nil {
errs = append(errs, "health block required for Service (Traefik routing requires health checks)")
}
if spec.Service != nil {
if err := validateServiceBind(spec.Service.Bind); err != nil {
errs = append(errs, err.Error())
}
}
errs = append(errs, validateTaskGroup(spec)...)
return composeErrors("schema/Service", errs)
}
// validateServiceBind validates the service.bind field (R-007). Empty
// is OK (default = socket). When set, it must be a valid IPv4/IPv6
// address (the only opt-in to bind on a non-loopback address); the
// loopback 127.0.0.1 is the documented opt-in. Anything that is not
// parseable as an IP address is rejected.
func validateServiceBind(bind string) error {
if strings.TrimSpace(bind) == "" {
return nil
}
if net.ParseIP(bind) == nil {
return fmt.Errorf("service.bind %q is not a valid IP address (R-007: 127.0.0.1 opt-in; default is socket)", bind)
}
return nil
}
// Validate validates a DaemonSet spec. See DaemonSetValidator for the rules.
func (DaemonSetValidator) Validate(spec *jobspec.WorkloadSpec) error {
if spec == nil {
return errors.New("schema/DaemonSet: spec is nil")
}
var errs []string
if strings.TrimSpace(spec.Name) == "" {
errs = append(errs, "name is required")
}
if spec.Schedule == nil {
errs = append(errs, "schedule block required for DaemonSet")
} else {
switch spec.Schedule.Mode {
case "every-node", "matching", "mandatory":
case "":
errs = append(errs, "schedule mode required for DaemonSet (one of every-node, matching, mandatory)")
default:
errs = append(errs, fmt.Sprintf("schedule mode %q invalid (want one of every-node, matching, mandatory)", spec.Schedule.Mode))
}
}
if len(spec.Ports) > 0 {
errs = append(errs, "ports not allowed for DaemonSet (no Traefik route by default D-175)")
}
if spec.Count != 0 {
errs = append(errs, fmt.Sprintf("count not allowed for DaemonSet (implicit = nodes matching condition), got %d", spec.Count))
}
if spec.Restart == nil {
errs = append(errs, "restart block required for DaemonSet")
}
errs = append(errs, validateTaskGroup(spec)...)
return composeErrors("schema/DaemonSet", errs)
}
// composeErrors joins the per-field errors into a single error prefixed
// by the validator name. Returns nil when there are no errors so the
// caller can return the result directly.
func composeErrors(name string, errs []string) error {
if len(errs) == 0 {
return nil
}
return fmt.Errorf("%s: %s", name, strings.Join(errs, "; "))
}
// validateTaskGroup validates the task-group list shared by all kinds
// (P06, PRD §9.1). When the spec carries a task group (spec.Tasks
// non-empty), each task must have a unique name and a resolvable
// command (the task's own Command, the task's runtime command, or the
// top-level runtime command as the per-group default). The top-level
// runtime is optional when tasks is present (each task can carry its
// own runtime). Returns nil when the spec has no task group.
func validateTaskGroup(spec *jobspec.WorkloadSpec) []string {
if len(spec.Tasks) == 0 {
return nil
}
var errs []string
seen := make(map[string]bool, len(spec.Tasks))
for i, task := range spec.Tasks {
if strings.TrimSpace(task.Name) == "" {
errs = append(errs, fmt.Sprintf("tasks[%d]: name is required", i))
} else if seen[task.Name] {
errs = append(errs, fmt.Sprintf("tasks[%d]: duplicate task name %q (names must be unique within the group)", i, task.Name))
} else {
seen[task.Name] = true
}
cmd := task.Command
if strings.TrimSpace(cmd) == "" && task.Runtime != nil {
cmd = task.Runtime.Command
}
if strings.TrimSpace(cmd) == "" && spec.Runtime != nil {
cmd = spec.Runtime.Command
}
if strings.TrimSpace(cmd) == "" {
errs = append(errs, fmt.Sprintf("tasks[%d]: command is required (set tasks[].command, tasks[].runtime.command, or top-level runtime.command)", i))
}
}
return errs
}