From 60b0357eb65880699ff1f8bbe52140d0726434f4 Mon Sep 17 00:00:00 2001 From: Jon Chery Date: Wed, 5 Aug 2026 17:17:02 +0000 Subject: [PATCH] feat(P0c): Job/Service/DaemonSet schemas + emitter interface + systemd stub (REQ-074) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0c — Kind-specific schema validators + Layer 4 emitter interface. Schemas (internal/spec/schema/schema.go, REQ-074): - Validator interface with JobValidator, ServiceValidator, DaemonSetValidator. JobValidator: count=1, no service block, optional schedule/timeout. ServiceValidator: ports required, count>=1, restart+update+runtime required. DaemonSetValidator: schedule mode required, no ports (D-175), no count. ValidatorFor(kind) dispatcher. 96.2% coverage. Emitter interface (internal/emitter/emitter.go, REQ-074, I-B-002): - File{Path,Content,Mode}, Emitter interface { Render(spec,node) []File }, Registry keyed by kind:runtime, Register + Render lookup. 100% coverage. Systemd stub (internal/emitter/systemd.go): - SystemdEmitter for process runtime. Renders minimal [Service] unit at /etc/systemd/system/orca-v1-alloc-.service (orca-v1- prefix per dual-write window REQ-090 — no overlap with v0.8 daemon's orca-). Flock test fix: TestFlock_concurrentBlocks rewritten to use non-blocking tryFlockEx (LOCK_NB) instead of a leaked blocking goroutine. Eliminates the temp-dir cleanup race. 20 packages pass, 20 bats pass, gofmt clean, verify-reqs 90 consistent. ---ci--- project: orca phase: P0c milestone: v0.9 status: execute ---/ci--- --- internal/emitter/emitter.go | 94 ++++++ internal/emitter/emitter_test.go | 163 +++++++++++ internal/emitter/systemd.go | 60 ++++ internal/emitter/systemd_test.go | 156 ++++++++++ internal/jobspec/markdown.go | 62 ++++ internal/security/flock.go | 8 + internal/security/flock_test.go | 35 ++- internal/spec/schema/schema.go | 164 +++++++++++ internal/spec/schema/schema_test.go | 436 ++++++++++++++++++++++++++++ 9 files changed, 1171 insertions(+), 7 deletions(-) create mode 100644 internal/emitter/emitter.go create mode 100644 internal/emitter/emitter_test.go create mode 100644 internal/emitter/systemd.go create mode 100644 internal/emitter/systemd_test.go create mode 100644 internal/spec/schema/schema.go create mode 100644 internal/spec/schema/schema_test.go diff --git a/internal/emitter/emitter.go b/internal/emitter/emitter.go new file mode 100644 index 0000000..858d107 --- /dev/null +++ b/internal/emitter/emitter.go @@ -0,0 +1,94 @@ +// Package emitter defines the Layer-4 emitter interface (REQ-074, +// I-B-002): the bridge between the declarative *jobspec.WorkloadSpec +// and the server-side files. An Emitter renders a *WorkloadSpec into a +// slice of File artifacts that the SSH-push transport SCPs to peers. +// +// Emitters are registered per workload kind + runtime (e.g. +// "service:wasm", "job:process", "daemonset:wasm"). The Registry looks +// up the right emitter by ":" and delegates. Unknown +// combinations return an error so the caller can fail fast before any +// file is written. +// +// P0c only ships the interface, the Registry, a stub SystemdEmitter +// (process runtime), and the File/Node value types. The full emitter +// implementations (systemd lifecycle hooks, Traefik, Syncthing, +// sockets) land in later phases (P02 Traefik, P04 lifecycle, P08 +// sockets, P09 Syncthing, v0.10-P03 secrets). +package emitter + +import ( + "fmt" + "strings" + + "git.cloudinit.dev/coreci/orca/internal/jobspec" +) + +// File is a single rendered artifact destined for a peer. The SSH-push +// transport writes Content to Path atomically (write-to-tmp + rename) +// with the given Mode (an octal string like "0644"). +type File struct { + Path string + Content string + Mode string +} + +// Node is the minimal peer description an emitter needs to render +// node-specific paths. It carries the hostname, the runtimes available +// on the node (so emitters can branch), and the node tags (used by +// DaemonSet matching and affinity in P05). +type Node struct { + Hostname string + Runtime []string + Tags []string +} + +// Emitter renders a *jobspec.WorkloadSpec for a given Node into a slice +// of File artifacts. Implementations are registered with a Registry +// keyed by ":". +type Emitter interface { + Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error) +} + +// Registry holds emitters keyed by ":" (e.g. +// "service:process", "job:wasm"). The zero-value Registry is not +// usable; construct one with NewRegistry. +type Registry struct { + emitters map[string]Emitter +} + +// NewRegistry returns an empty Registry ready for Register calls. +func NewRegistry() *Registry { + return &Registry{emitters: make(map[string]Emitter)} +} + +// Register registers an Emitter under the given key. The key is +// ":" (e.g. "job:process"). Registering twice under the +// same key overwrites the prior registration (last-wins) to keep the +// surface simple; callers are responsible for not double-registering. +func (r *Registry) Register(key string, e Emitter) { + r.emitters[key] = e +} + +// Render looks up the emitter for ":" in the registry and +// delegates to it. The kind is lowercased so the canonical spec kinds +// (Job, Service, DaemonSet) map to the lowercase registry keys +// ("job:process", "service:wasm", "daemonset:process"). Returns an error +// if the spec is nil, the spec is missing its Kind, the runtime is +// missing, or no emitter is registered for the combination. +func (r *Registry) Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error) { + if spec == nil { + return nil, fmt.Errorf("emitter: spec is nil") + } + if strings.TrimSpace(spec.Kind) == "" { + return nil, fmt.Errorf("emitter: spec kind is empty") + } + if spec.Runtime == nil { + return nil, fmt.Errorf("emitter: spec runtime is nil") + } + key := strings.ToLower(spec.Kind) + ":" + spec.Runtime.OneOf + e, ok := r.emitters[key] + if !ok { + return nil, fmt.Errorf("emitter: no emitter registered for %q (kind:runtime)", key) + } + return e.Render(spec, node) +} diff --git a/internal/emitter/emitter_test.go b/internal/emitter/emitter_test.go new file mode 100644 index 0000000..7a77e7c --- /dev/null +++ b/internal/emitter/emitter_test.go @@ -0,0 +1,163 @@ +package emitter + +import ( + "errors" + "strings" + "testing" + + "git.cloudinit.dev/coreci/orca/internal/jobspec" +) + +// mockEmitter is a test-only Emitter that returns a fixed File slice +// (or an error) so the Registry tests do not depend on the +// SystemdEmitter. Implements Emitter via value receiver. +type mockEmitter struct { + files []File + err error +} + +func (m mockEmitter) Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error) { + if m.err != nil { + return nil, m.err + } + out := make([]File, len(m.files)) + copy(out, m.files) + return out, nil +} + +func TestRegistry_RegisterAndRender(t *testing.T) { + r := NewRegistry() + want := []File{{Path: "/tmp/a", Content: "alpha", Mode: "0644"}} + r.Register("job:process", mockEmitter{files: want}) + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "demo", + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/true"}, + } + node := &Node{Hostname: "node-1", Runtime: []string{"process"}} + got, err := r.Render(spec, node) + if err != nil { + t.Fatalf("Render: %v", err) + } + if len(got) != 1 { + t.Fatalf("got %d files, want 1", len(got)) + } + if got[0] != want[0] { + t.Errorf("file = %+v, want %+v", got[0], want[0]) + } +} + +func TestRegistry_UnknownKindRuntime(t *testing.T) { + r := NewRegistry() + spec := &jobspec.WorkloadSpec{ + Kind: "Service", + Name: "web", + Runtime: &jobspec.RuntimeBlock{OneOf: "wasm"}, + } + _, err := r.Render(spec, &Node{}) + if err == nil { + t.Fatal("expected error for unknown kind:runtime, got nil") + } + if !strings.Contains(err.Error(), "no emitter registered") { + t.Errorf("error = %q, want 'no emitter registered'", err.Error()) + } + if !strings.Contains(err.Error(), "service:wasm") { + t.Errorf("error = %q, want it to mention 'service:wasm'", err.Error()) + } +} + +func TestRegistry_MultipleEmittersCorrectSelected(t *testing.T) { + r := NewRegistry() + jobFiles := []File{{Path: "/tmp/job", Content: "job", Mode: "0644"}} + svcFiles := []File{{Path: "/tmp/svc", Content: "svc", Mode: "0644"}} + dsFiles := []File{{Path: "/tmp/ds", Content: "ds", Mode: "0644"}} + r.Register("job:process", mockEmitter{files: jobFiles}) + r.Register("service:process", mockEmitter{files: svcFiles}) + r.Register("daemonset:process", mockEmitter{files: dsFiles}) + + cases := []struct { + kind string + runtime string + wantPath string + }{ + {"Job", "process", "/tmp/job"}, + {"Service", "process", "/tmp/svc"}, + {"DaemonSet", "process", "/tmp/ds"}, + } + for _, tc := range cases { + t.Run(tc.kind+":"+tc.runtime, func(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: tc.kind, + Name: "x", + Runtime: &jobspec.RuntimeBlock{OneOf: tc.runtime, Command: "/bin/x"}, + } + got, err := r.Render(spec, &Node{Hostname: "n"}) + if err != nil { + t.Fatalf("Render: %v", err) + } + if len(got) != 1 { + t.Fatalf("got %d files, want 1", len(got)) + } + if got[0].Path != tc.wantPath { + t.Errorf("path = %q, want %q", got[0].Path, tc.wantPath) + } + }) + } +} + +func TestRegistry_NilSpec(t *testing.T) { + r := NewRegistry() + _, err := r.Render(nil, &Node{}) + if err == nil { + t.Fatal("expected error for nil spec") + } + if !strings.Contains(err.Error(), "spec is nil") { + t.Errorf("error = %q, want 'spec is nil'", err.Error()) + } +} + +func TestRegistry_EmptyKind(t *testing.T) { + r := NewRegistry() + spec := &jobspec.WorkloadSpec{Runtime: &jobspec.RuntimeBlock{OneOf: "process"}} + _, err := r.Render(spec, &Node{}) + if err == nil { + t.Fatal("expected error for empty kind") + } + if !strings.Contains(err.Error(), "kind is empty") { + t.Errorf("error = %q, want 'kind is empty'", err.Error()) + } +} + +func TestRegistry_NilRuntime(t *testing.T) { + r := NewRegistry() + spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x"} + _, err := r.Render(spec, &Node{}) + if err == nil { + t.Fatal("expected error for nil runtime") + } + if !strings.Contains(err.Error(), "runtime is nil") { + t.Errorf("error = %q, want 'runtime is nil'", err.Error()) + } +} + +func TestRegistry_EmitterErrorPropagates(t *testing.T) { + r := NewRegistry() + wantErr := errors.New("boom") + r.Register("job:process", mockEmitter{err: wantErr}) + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "x", + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/x"}, + } + _, err := r.Render(spec, &Node{}) + if !errors.Is(err, wantErr) { + t.Errorf("err = %v, want %v", err, wantErr) + } +} + +// Compile-time assertion that mockEmitter and SystemdEmitter implement +// Emitter. +var ( + _ Emitter = mockEmitter{} + _ Emitter = SystemdEmitter{} +) diff --git a/internal/emitter/systemd.go b/internal/emitter/systemd.go new file mode 100644 index 0000000..4956b74 --- /dev/null +++ b/internal/emitter/systemd.go @@ -0,0 +1,60 @@ +package emitter + +import ( + "errors" + "fmt" + "strings" + + "git.cloudinit.dev/coreci/orca/internal/jobspec" +) + +// SystemdEmitter is a stub Emitter implementation for the "process" +// runtime. It renders a minimal systemd unit file for the workload. +// +// This is a STUB — the full systemd emitter (with lifecycle hooks, +// sockets, EnvironmentFile, LoadCredential) lands in later phases: +// +// - P04: lifecycle hooks (ExecStop, ExecStartPre/Post, timeouts) +// - P08: socket plumbing (R-007) +// - v0.10-P03: secrets via EnvironmentFile= + LoadCredential= +// +// P0c ships only the minimal [Service]\nExecStart=... shape to prove +// the Emitter interface end-to-end. The unit name carries the +// `orca-v1-` prefix per the dual-write window (REQ-090) so the v0.9 +// SSH-push path does not collide with the v0.8 daemon's +// `orca-.service` units during the migration window. +type SystemdEmitter struct{} + +// unitNamePrefix is the v0.9 SSH-push unit-name prefix. The v0.8 +// daemon uses `orca-.service`; the v0.9 path uses +// `orca-v1-.service` so the two never overlap (REQ-090, +// I-C-006). The prefix is load-bearing — do not change it without +// updating the dual-write window contract. +const unitNamePrefix = "orca-v1-" + +// Render renders a minimal systemd unit file for a process-runtime +// workload. The unit name is `/etc/systemd/system/.service` +// and the content is a minimal `[Service]` block with the runtime +// command as ExecStart. Mode is 0644 (the lead applier chmods after +// atomic rename). +// +// Returns an error if the spec is nil, the spec is missing its name, +// or the runtime command is empty (a workload with no command has +// nothing to ExecStart). +func (SystemdEmitter) Render(spec *jobspec.WorkloadSpec, node *Node) ([]File, error) { + if spec == nil { + return nil, errors.New("emitter/systemd: spec is nil") + } + if strings.TrimSpace(spec.Name) == "" { + return nil, errors.New("emitter/systemd: spec name is empty") + } + if spec.Runtime == nil { + return nil, errors.New("emitter/systemd: runtime block is nil") + } + if strings.TrimSpace(spec.Runtime.Command) == "" { + return nil, errors.New("emitter/systemd: runtime command is empty") + } + path := fmt.Sprintf("/etc/systemd/system/%s%s.service", unitNamePrefix, spec.Name) + content := fmt.Sprintf("[Service]\nExecStart=%s\n", spec.Runtime.Command) + return []File{{Path: path, Content: content, Mode: "0644"}}, nil +} diff --git a/internal/emitter/systemd_test.go b/internal/emitter/systemd_test.go new file mode 100644 index 0000000..ee6f162 --- /dev/null +++ b/internal/emitter/systemd_test.go @@ -0,0 +1,156 @@ +package emitter + +import ( + "strings" + "testing" + + "git.cloudinit.dev/coreci/orca/internal/jobspec" +) + +func TestSystemdEmitter_RenderJob(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "backup", + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/usr/bin/rsync -a /src /dst"}, + } + node := &Node{Hostname: "node-1", Runtime: []string{"process"}} + files, err := SystemdEmitter{}.Render(spec, node) + if err != nil { + t.Fatalf("Render: %v", err) + } + if len(files) != 1 { + t.Fatalf("got %d files, want 1", len(files)) + } + f := files[0] + wantPath := "/etc/systemd/system/orca-v1-backup.service" + if f.Path != wantPath { + t.Errorf("Path = %q, want %q", f.Path, wantPath) + } + wantContent := "[Service]\nExecStart=/usr/bin/rsync -a /src /dst\n" + if f.Content != wantContent { + t.Errorf("Content = %q, want %q", f.Content, wantContent) + } + if f.Mode != "0644" { + t.Errorf("Mode = %q, want 0644", f.Mode) + } +} + +func TestSystemdEmitter_RenderService(t *testing.T) { + // The full service emitter (Traefik route + health checks) lands in + // P02; here we only prove the systemd side renders for a Service + // kind with a process runtime. + spec := &jobspec.WorkloadSpec{ + Kind: "Service", + Name: "web", + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/usr/local/bin/httpd -f"}, + } + node := &Node{Hostname: "node-1", Runtime: []string{"process"}} + files, err := SystemdEmitter{}.Render(spec, node) + if err != nil { + t.Fatalf("Render: %v", err) + } + if len(files) != 1 { + t.Fatalf("got %d files, want 1", len(files)) + } + if files[0].Path != "/etc/systemd/system/orca-v1-web.service" { + t.Errorf("Path = %q, want /etc/systemd/system/orca-v1-web.service", files[0].Path) + } + if !strings.Contains(files[0].Content, "ExecStart=/usr/local/bin/httpd -f") { + t.Errorf("Content = %q, want it to contain the ExecStart line", files[0].Content) + } +} + +func TestSystemdEmitter_EmptyCommandError(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "x", + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: ""}, + } + _, err := SystemdEmitter{}.Render(spec, &Node{}) + if err == nil { + t.Fatal("expected error for empty command, got nil") + } + if !strings.Contains(err.Error(), "command is empty") { + t.Errorf("error = %q, want 'command is empty'", err.Error()) + } +} + +func TestSystemdEmitter_WhitespaceCommandError(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "x", + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: " "}, + } + _, err := SystemdEmitter{}.Render(spec, &Node{}) + if err == nil { + t.Fatal("expected error for whitespace-only command, got nil") + } + if !strings.Contains(err.Error(), "command is empty") { + t.Errorf("error = %q, want 'command is empty'", err.Error()) + } +} + +func TestSystemdEmitter_NilSpec(t *testing.T) { + v := SystemdEmitter{} + if _, err := v.Render(nil, &Node{}); err == nil { + t.Fatal("expected error for nil spec") + } +} + +func TestSystemdEmitter_EmptyName(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: " ", + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/x"}, + } + _, err := SystemdEmitter{}.Render(spec, &Node{}) + if err == nil { + t.Fatal("expected error for empty name") + } + if !strings.Contains(err.Error(), "name is empty") { + t.Errorf("error = %q, want 'name is empty'", err.Error()) + } +} + +func TestSystemdEmitter_NilRuntime(t *testing.T) { + spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "x"} + _, err := SystemdEmitter{}.Render(spec, &Node{}) + if err == nil { + t.Fatal("expected error for nil runtime") + } + if !strings.Contains(err.Error(), "runtime block is nil") { + t.Errorf("error = %q, want 'runtime block is nil'", err.Error()) + } +} + +func TestSystemdEmitter_UnitNamePrefix(t *testing.T) { + // The orca-v1- prefix is load-bearing for the dual-write window + // (REQ-090, I-C-006): the v0.8 daemon writes `orca-.service` + // and the v0.9 SSH-push path writes `orca-v1-.service`, + // so the two never collide. This test guards against accidental + // removal of the prefix. + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "dual-write-safety", + 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.HasPrefix(files[0].Path, "/etc/systemd/system/orca-v1-") { + t.Errorf("Path = %q, want it to start with /etc/systemd/system/orca-v1- (REQ-090)", files[0].Path) + } + if !strings.HasSuffix(files[0].Path, ".service") { + t.Errorf("Path = %q, want it to end with .service", files[0].Path) + } + // Explicitly assert the full expected unit name to lock the contract. + want := "/etc/systemd/system/orca-v1-dual-write-safety.service" + if files[0].Path != want { + t.Errorf("Path = %q, want %q", files[0].Path, want) + } + // Sanity: the prefix is exactly "orca-v1-", not "orca-v0" or "orca". + if unitNamePrefix != "orca-v1-" { + t.Errorf("unitNamePrefix = %q, want orca-v1-", unitNamePrefix) + } +} diff --git a/internal/jobspec/markdown.go b/internal/jobspec/markdown.go index e1f346e..a62b437 100644 --- a/internal/jobspec/markdown.go +++ b/internal/jobspec/markdown.go @@ -24,6 +24,37 @@ type WorkloadSpec struct { Secrets []string Volumes []VolumeSpec Body string + + // Kind-specific blocks consumed by the P0c schema validators + // (internal/spec/schema). The Markdown parser does not populate + // these yet; later phases (P02 service block, P03 update stanza, + // P04 lifecycle hooks) extend the parser. P0c only defines the + // struct shape so validators can reference the fields. + + // Restart is the restart policy block. Required for Service and + // DaemonSet; optional for Job (defaults to never/on-failure). + // Populated by the P04 lifecycle phase. + Restart *RestartBlock + + // Schedule is the schedule block. For Job it carries an optional + // cron string; for DaemonSet it carries the placement mode + // (every-node/matching/mandatory). Populated by P05 (scheduler + // skeleton) and the DaemonSet phase. + Schedule *ScheduleBlock + + // Update is the rolling/canary update stanza. Required for + // Service. Populated by the P03 update-stanza phase. + Update *UpdateBlock + + // Service is the service block (Traefik route definition). For + // Service kind it is implied; Job and DaemonSet do not carry a + // Traefik route by default (D-175). Populated by the P02 service + // block phase. + Service *ServiceBlock + + // Timeout is an optional execution timeout (duration string) for + // Job. Populated by P04. + Timeout string } // RuntimeBlock is a minimal runtime abstraction surface populated by the @@ -35,6 +66,37 @@ type RuntimeBlock struct { Command string } +// RestartBlock is the restart policy block. Mode is one of never, +// on-failure, service (REQ-074 schema validators). Populated by P04. +type RestartBlock struct { + Mode string + MaxRetries int + Delay string +} + +// ScheduleBlock is the scheduling block. For Job, Cron is an optional +// cron expression. For DaemonSet, Mode is one of every-node, matching, +// mandatory (REQ-074). Populated by P05 and the DaemonSet phase. +type ScheduleBlock struct { + Mode string + Cron string +} + +// UpdateBlock is the rolling/canary update stanza. Required for Service. +// Populated by P03. +type UpdateBlock struct { + Strategy string + MaxSurge int +} + +// ServiceBlock is the Traefik route definition. For Service it is +// implied (Traefik route YES); Job and DaemonSet do not carry one by +// default (D-175). Populated by P02. +type ServiceBlock struct { + Host string + RouteID string +} + // PortSpec is a minimal port binding entry. HostIP is optional. type PortSpec struct { Name string diff --git a/internal/security/flock.go b/internal/security/flock.go index df33258..78552a9 100644 --- a/internal/security/flock.go +++ b/internal/security/flock.go @@ -25,3 +25,11 @@ func Flock(path string) (release func(), err error) { _ = f.Close() }, nil } + +func tryFlockEx(fd int) error { + return syscall.Flock(fd, syscall.LOCK_EX|syscall.LOCK_NB) +} + +func releaseFlock(fd int) error { + return syscall.Flock(fd, syscall.LOCK_UN) +} diff --git a/internal/security/flock_test.go b/internal/security/flock_test.go index f44bcc5..84dc112 100644 --- a/internal/security/flock_test.go +++ b/internal/security/flock_test.go @@ -4,6 +4,7 @@ import ( "os" "path/filepath" "testing" + "time" ) func TestFlock_acquireAndRelease(t *testing.T) { @@ -45,17 +46,37 @@ func TestFlock_concurrentBlocks(t *testing.T) { if err != nil { t.Fatalf("first Flock: %v", err) } - defer r1() - done := make(chan error, 1) + // Give the blocking goroutine a chance to start and block. + time.Sleep(50 * time.Millisecond) + + // Verify the second lock is blocked by checking it hasn't acquired after a short window. + // Use a non-blocking attempt: open the file and try LOCK_EX|LOCK_NB. + blocked := make(chan bool, 1) go func() { - _, err := Flock(path) - done <- err + f, err := os.OpenFile(path, os.O_RDWR, 0o600) + if err != nil { + blocked <- false + return + } + defer f.Close() + // LOCK_NB = non-blocking; returns EWOULDBLOCK if locked. + if err := tryFlockEx(int(f.Fd())); err != nil { + blocked <- true // got EWOULDBLOCK = the lock is held by r1 + return + } + releaseFlock(int(f.Fd())) + blocked <- false // acquired = r1 didn't hold the lock (bug) }() select { - case <-done: - t.Fatal("second Flock should block while first holds the lock") - default: + case b := <-blocked: + if !b { + t.Fatal("second lock acquired while first holds it — lock not working") + } + case <-time.After(2 * time.Second): + t.Fatal("non-blocking try-lock timed out") } + + r1() } diff --git a/internal/spec/schema/schema.go b/internal/spec/schema/schema.go new file mode 100644 index 0000000..571c88b --- /dev/null +++ b/internal/spec/schema/schema.go @@ -0,0 +1,164 @@ +// 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" + "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) +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 +// - runtime required +// - service block implied (Traefik route YES) +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 +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)") + } + 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 if spec.Restart.Mode != "service" { + errs = append(errs, fmt.Sprintf("restart mode must be %q for Service, got %q", "service", spec.Restart.Mode)) + } + if spec.Update == nil { + errs = append(errs, "update block required for Service") + } + if spec.Runtime == nil { + errs = append(errs, "runtime block required for Service") + } + return composeErrors("schema/Service", errs) +} + +// 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") + } + 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, "; ")) +} diff --git a/internal/spec/schema/schema_test.go b/internal/spec/schema/schema_test.go new file mode 100644 index 0000000..47492b9 --- /dev/null +++ b/internal/spec/schema/schema_test.go @@ -0,0 +1,436 @@ +package schema + +import ( + "strings" + "testing" + + "git.cloudinit.dev/coreci/orca/internal/jobspec" +) + +func TestJobValidator_ValidMinimal(t *testing.T) { + spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "backup", Count: 1} + v := JobValidator{} + if err := v.Validate(spec); err != nil { + t.Fatalf("expected nil, got %v", err) + } +} + +func TestJobValidator_ValidWithSchedule(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "backup", + Count: 1, + Schedule: &jobspec.ScheduleBlock{Cron: "0 2 * * *"}, + Timeout: "1h", + } + v := JobValidator{} + if err := v.Validate(spec); err != nil { + t.Fatalf("expected nil, got %v", err) + } +} + +func TestJobValidator_ValidUnsetCount(t *testing.T) { + spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "one-shot"} + v := JobValidator{} + if err := v.Validate(spec); err != nil { + t.Fatalf("unset count should default-accept, got %v", err) + } +} + +func TestJobValidator_MissingName(t *testing.T) { + spec := &jobspec.WorkloadSpec{Kind: "Job", Count: 1} + err := JobValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for missing name, got nil") + } + if !strings.Contains(err.Error(), "name is required") { + t.Errorf("error = %q, want 'name is required'", err.Error()) + } +} + +func TestJobValidator_CountGreaterThanOne(t *testing.T) { + spec := &jobspec.WorkloadSpec{Kind: "Job", Name: "batch", Count: 3} + err := JobValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for count > 1, got nil") + } + if !strings.Contains(err.Error(), "count must be 1") { + t.Errorf("error = %q, want 'count must be 1'", err.Error()) + } +} + +func TestJobValidator_ServiceBlockRejected(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "x", + Count: 1, + Service: &jobspec.ServiceBlock{Host: "x.example"}, + } + err := JobValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for service block on Job, got nil") + } + if !strings.Contains(err.Error(), "service block") { + t.Errorf("error = %q, want 'service block'", err.Error()) + } +} + +func TestJobValidator_NilSpec(t *testing.T) { + v := JobValidator{} + if err := v.Validate(nil); err == nil { + t.Fatal("expected error for nil spec") + } +} + +func TestServiceValidator_ValidFull(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Service", + Name: "web", + Count: 3, + Runtime: &jobspec.RuntimeBlock{OneOf: "process", Command: "/bin/http"}, + Restart: &jobspec.RestartBlock{Mode: "service"}, + Update: &jobspec.UpdateBlock{Strategy: "rolling", MaxSurge: 1}, + Ports: []jobspec.PortSpec{{Name: "http", Port: 8080}}, + } + v := ServiceValidator{} + if err := v.Validate(spec); err != nil { + t.Fatalf("expected nil, got %v", err) + } +} + +func TestServiceValidator_MissingPorts(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Service", + Name: "web", + Count: 2, + Runtime: &jobspec.RuntimeBlock{OneOf: "process"}, + Restart: &jobspec.RestartBlock{Mode: "service"}, + Update: &jobspec.UpdateBlock{Strategy: "rolling"}, + } + err := ServiceValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for missing ports, got nil") + } + if !strings.Contains(err.Error(), "ports required") { + t.Errorf("error = %q, want 'ports required'", err.Error()) + } +} + +func TestServiceValidator_CountZero(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Service", + Name: "web", + Count: 0, + Runtime: &jobspec.RuntimeBlock{OneOf: "process"}, + Restart: &jobspec.RestartBlock{Mode: "service"}, + Update: &jobspec.UpdateBlock{Strategy: "rolling"}, + Ports: []jobspec.PortSpec{{Name: "http", Port: 8080}}, + } + err := ServiceValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for count 0, got nil") + } + if !strings.Contains(err.Error(), "count must be") { + t.Errorf("error = %q, want 'count must be'", err.Error()) + } +} + +func TestServiceValidator_MissingRestart(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Service", + Name: "web", + Count: 1, + Runtime: &jobspec.RuntimeBlock{OneOf: "process"}, + Update: &jobspec.UpdateBlock{Strategy: "rolling"}, + Ports: []jobspec.PortSpec{{Name: "http", Port: 8080}}, + } + err := ServiceValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for missing restart, got nil") + } + if !strings.Contains(err.Error(), "restart block required") { + t.Errorf("error = %q, want 'restart block required'", err.Error()) + } +} + +func TestServiceValidator_WrongRestartMode(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Service", + Name: "web", + Count: 1, + Runtime: &jobspec.RuntimeBlock{OneOf: "process"}, + Restart: &jobspec.RestartBlock{Mode: "on-failure"}, + Update: &jobspec.UpdateBlock{Strategy: "rolling"}, + Ports: []jobspec.PortSpec{{Name: "http", Port: 8080}}, + } + err := ServiceValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for wrong restart mode, got nil") + } + if !strings.Contains(err.Error(), "restart mode must be") { + t.Errorf("error = %q, want 'restart mode must be'", err.Error()) + } +} + +func TestServiceValidator_MissingUpdate(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Service", + Name: "web", + Count: 1, + Runtime: &jobspec.RuntimeBlock{OneOf: "process"}, + Restart: &jobspec.RestartBlock{Mode: "service"}, + Ports: []jobspec.PortSpec{{Name: "http", Port: 8080}}, + } + err := ServiceValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for missing update, got nil") + } + if !strings.Contains(err.Error(), "update block required") { + t.Errorf("error = %q, want 'update block required'", err.Error()) + } +} + +func TestServiceValidator_MissingRuntime(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "Service", + Name: "web", + Count: 1, + Restart: &jobspec.RestartBlock{Mode: "service"}, + Update: &jobspec.UpdateBlock{Strategy: "rolling"}, + Ports: []jobspec.PortSpec{{Name: "http", Port: 8080}}, + } + err := ServiceValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for missing runtime, got nil") + } + if !strings.Contains(err.Error(), "runtime block required") { + t.Errorf("error = %q, want 'runtime block required'", err.Error()) + } +} + +func TestServiceValidator_NilSpec(t *testing.T) { + v := ServiceValidator{} + if err := v.Validate(nil); err == nil { + t.Fatal("expected error for nil spec") + } +} + +func TestDaemonSetValidator_Valid(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "DaemonSet", + Name: "log-shipper", + Schedule: &jobspec.ScheduleBlock{Mode: "every-node"}, + Restart: &jobspec.RestartBlock{Mode: "on-failure"}, + } + v := DaemonSetValidator{} + if err := v.Validate(spec); err != nil { + t.Fatalf("expected nil, got %v", err) + } +} + +func TestDaemonSetValidator_ValidMatchingMode(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "DaemonSet", + Name: "x", + Schedule: &jobspec.ScheduleBlock{Mode: "matching"}, + Restart: &jobspec.RestartBlock{Mode: "on-failure"}, + } + v := DaemonSetValidator{} + if err := v.Validate(spec); err != nil { + t.Fatalf("matching mode should be accepted, got %v", err) + } +} + +func TestDaemonSetValidator_ValidMandatoryMode(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "DaemonSet", + Name: "x", + Schedule: &jobspec.ScheduleBlock{Mode: "mandatory"}, + Restart: &jobspec.RestartBlock{Mode: "on-failure"}, + } + v := DaemonSetValidator{} + if err := v.Validate(spec); err != nil { + t.Fatalf("mandatory mode should be accepted, got %v", err) + } +} + +func TestDaemonSetValidator_MissingScheduleMode(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "DaemonSet", + Name: "x", + Schedule: &jobspec.ScheduleBlock{Mode: ""}, + Restart: &jobspec.RestartBlock{Mode: "on-failure"}, + } + err := DaemonSetValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for missing schedule mode, got nil") + } + if !strings.Contains(err.Error(), "schedule mode required") { + t.Errorf("error = %q, want 'schedule mode required'", err.Error()) + } +} + +func TestDaemonSetValidator_MissingScheduleBlock(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "DaemonSet", + Name: "x", + Restart: &jobspec.RestartBlock{Mode: "on-failure"}, + } + err := DaemonSetValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for missing schedule block, got nil") + } + if !strings.Contains(err.Error(), "schedule block required") { + t.Errorf("error = %q, want 'schedule block required'", err.Error()) + } +} + +func TestDaemonSetValidator_InvalidScheduleMode(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "DaemonSet", + Name: "x", + Schedule: &jobspec.ScheduleBlock{Mode: "always"}, + Restart: &jobspec.RestartBlock{Mode: "on-failure"}, + } + err := DaemonSetValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for invalid schedule mode, got nil") + } + if !strings.Contains(err.Error(), "schedule mode") { + t.Errorf("error = %q, want 'schedule mode'", err.Error()) + } +} + +func TestDaemonSetValidator_HasPorts(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "DaemonSet", + Name: "x", + Schedule: &jobspec.ScheduleBlock{Mode: "every-node"}, + Restart: &jobspec.RestartBlock{Mode: "on-failure"}, + Ports: []jobspec.PortSpec{{Name: "http", Port: 80}}, + } + err := DaemonSetValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for ports on DaemonSet, got nil") + } + if !strings.Contains(err.Error(), "ports not allowed") { + t.Errorf("error = %q, want 'ports not allowed'", err.Error()) + } +} + +func TestDaemonSetValidator_HasCount(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "DaemonSet", + Name: "x", + Count: 3, + Schedule: &jobspec.ScheduleBlock{Mode: "every-node"}, + Restart: &jobspec.RestartBlock{Mode: "on-failure"}, + } + err := DaemonSetValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for count on DaemonSet, got nil") + } + if !strings.Contains(err.Error(), "count not allowed") { + t.Errorf("error = %q, want 'count not allowed'", err.Error()) + } +} + +func TestDaemonSetValidator_MissingRestart(t *testing.T) { + spec := &jobspec.WorkloadSpec{ + Kind: "DaemonSet", + Name: "x", + Schedule: &jobspec.ScheduleBlock{Mode: "every-node"}, + } + err := DaemonSetValidator{}.Validate(spec) + if err == nil { + t.Fatal("expected error for missing restart on DaemonSet, got nil") + } + if !strings.Contains(err.Error(), "restart block required") { + t.Errorf("error = %q, want 'restart block required'", err.Error()) + } +} + +func TestDaemonSetValidator_NilSpec(t *testing.T) { + v := DaemonSetValidator{} + if err := v.Validate(nil); err == nil { + t.Fatal("expected error for nil spec") + } +} + +func TestValidatorFor_EachKind(t *testing.T) { + cases := []struct { + kind string + want string + }{ + {"Job", "schema.JobValidator"}, + {"Service", "schema.ServiceValidator"}, + {"DaemonSet", "schema.DaemonSetValidator"}, + } + for _, tc := range cases { + t.Run(tc.kind, func(t *testing.T) { + v, err := ValidatorFor(tc.kind) + if err != nil { + t.Fatalf("ValidatorFor(%q): %v", tc.kind, err) + } + got := fmtType(v) + if got != tc.want { + t.Errorf("ValidatorFor(%q) type = %q, want %q", tc.kind, got, tc.want) + } + }) + } +} + +func TestValidatorFor_UnknownKind(t *testing.T) { + _, err := ValidatorFor("CronJob") + if err == nil { + t.Fatal("expected error for unknown kind, got nil") + } + if !strings.Contains(err.Error(), "unknown kind") { + t.Errorf("error = %q, want 'unknown kind'", err.Error()) + } +} + +// fmtType returns a readable type name for a validator. Uses fmt.Sprintf +// with %T rather than reflection to keep the test surface minimal. +func fmtType(v Validator) string { + switch v.(type) { + case JobValidator: + return "schema.JobValidator" + case ServiceValidator: + return "schema.ServiceValidator" + case DaemonSetValidator: + return "schema.DaemonSetValidator" + default: + return "unknown" + } +} + +// Ensure composeErrors returns nil for empty input (covers the +// short-circuit branch that the validators rely on). +func TestComposeErrors_Empty(t *testing.T) { + if err := composeErrors("schema/X", nil); err != nil { + t.Errorf("composeErrors(nil) = %v, want nil", err) + } + if err := composeErrors("schema/X", []string{}); err != nil { + t.Errorf("composeErrors([]) = %v, want nil", err) + } +} + +// Ensure the error type returned by composeErrors is a non-nil error +// when violations are present (guards against accidental nil-return). +func TestComposeErrors_NonEmpty(t *testing.T) { + err := composeErrors("schema/X", []string{"a", "b"}) + if err == nil { + t.Fatal("expected non-nil error") + } + if !strings.Contains(err.Error(), "a") || !strings.Contains(err.Error(), "b") { + t.Errorf("error = %q, want both 'a' and 'b'", err.Error()) + } +} + +// Compile-time assertion that the validators implement the interface. +var ( + _ Validator = JobValidator{} + _ Validator = ServiceValidator{} + _ Validator = DaemonSetValidator{} +)