From c10779873b6aeaa829ed445113726cd6562db3de Mon Sep 17 00:00:00 2001 From: Jon Chery Date: Wed, 5 Aug 2026 18:02:51 +0000 Subject: [PATCH] feat(P05): CLI-side scheduler + CEL constraints + affinity (REQ-083) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P05 — Scheduler moves from daemon-side to CLI-side (R-001) with runtime-awareness. Scheduler (internal/scheduler/scheduler.go, REQ-083): - Pure Schedule(nodes, req) -> []Placement. Job=1 best-fit, Service=count replicas (anti-affinity default, colocation permitted), DaemonSet=1 per matching node. Score(node, req) = (FreeCPU*1000 + FreeMem); fits checks runtime compat (wasm->wasmtime, pve-vm/ct->proxmox), constraints (CEL AND), capacity. Affinity scoring (target + weight, anti-affinity for spreading). CEL evaluator (internal/scheduler/cel.go): - Hand-rolled recursive-descent (no CEL dep in go.mod). Subset: node.* attrs, literals, ==/!=/>=/<=/>, in/not in, and/or/not, parens. Anything outside subset returns error (no silent wrong answer). Schedule treats eval errors as non-fit (node skipped). 23 packages pass, 20 bats pass, gofmt clean, verify-reqs 90 consistent. 89.5% coverage on internal/scheduler. ---ci--- project: orca phase: P05 milestone: v0.9 status: execute ---/ci--- --- internal/scheduler/cel.go | 536 +++++++++++++++++++++++++++ internal/scheduler/cel_test.go | 210 +++++++++++ internal/scheduler/scheduler.go | 472 +++++++++++++++++++++++ internal/scheduler/scheduler_test.go | 451 ++++++++++++++++++++++ 4 files changed, 1669 insertions(+) create mode 100644 internal/scheduler/cel.go create mode 100644 internal/scheduler/cel_test.go create mode 100644 internal/scheduler/scheduler.go create mode 100644 internal/scheduler/scheduler_test.go diff --git a/internal/scheduler/cel.go b/internal/scheduler/cel.go new file mode 100644 index 0000000..2a26b0d --- /dev/null +++ b/internal/scheduler/cel.go @@ -0,0 +1,536 @@ +// Package scheduler — cel.go implements a minimal CEL-subset evaluator +// for the CLI-side scheduler constraint expressions (REQ-083, P05). +// +// The full CEL specification (google.golang.org/genproto/... +// googleapis/api/expr/v1alpha1) is intentionally NOT a dependency of +// this module (see go.mod): adding it for a single callsite would pull +// in a large transitive graph and contradict the "stdlib + minimal +// deps" guardrail. Instead this file implements a hand-rolled +// recursive-descent evaluator for the subset the PRD exercises: +// +// - attribute access on a `node.` object (hostname, kind, +// cpus, memory, tags, runtimes) +// - string and integer literals (double-quoted) +// - comparison operators: == != >= <= > < +// - membership: in , not in +// - boolean composition: and, or, not (parenthesised) +// +// Anything outside this subset returns an error rather than a silent +// wrong answer; that is the documented limitation. The grammar is +// small enough to be unambiguous with a top-down precedence-climbing +// parser. +package scheduler + +import ( + "fmt" + "strconv" + "strings" + "unicode" +) + +// EvaluateConstraint evaluates a single CEL-subset expression against +// the supplied NodeInfo. Returns (matched, err). An expression that +// references an unknown attribute, uses an unsupported operator, or +// fails to parse yields an error. Schedule treats a constraint +// evaluation error as a non-fit (the node is silently skipped) rather +// than a hard fail because operators routinely write exploratory +// constraints against attributes the local cluster does not expose. +func EvaluateConstraint(expr string, node NodeInfo) (bool, error) { + p := newParser(strings.TrimSpace(expr), node) + if p.len() == 0 { + return false, fmt.Errorf("cel: empty expression") + } + v, err := p.parseExpr() + if err != nil { + return false, err + } + if p.tok.kind != tokEOF { + return false, fmt.Errorf("cel: trailing input near %q", p.tok.text) + } + b, ok := v.(bool) + if !ok { + return false, fmt.Errorf("cel: expression did not evaluate to bool (got %T)", v) + } + return b, nil +} + +// EvaluateAll returns true iff every constraint evaluates to true +// against the node (logical AND). An empty constraint list is vacuously +// true. The first evaluation error short-circuits and is returned. +func EvaluateAll(constraints []string, node NodeInfo) (bool, error) { + for _, c := range constraints { + ok, err := EvaluateConstraint(c, node) + if err != nil { + return false, fmt.Errorf("constraint %q: %w", c, err) + } + if !ok { + return false, nil + } + } + return true, nil +} + +// ---------------------------------------------------------------------------- +// Value model +// ---------------------------------------------------------------------------- + +// celValue is the union of values the evaluator produces. We use the +// Go interface{} representation so that comparisons can be polymorphic +// without a tagged-union ceremony; the supported concrete types are +// bool, int64, and string. Lists are []celValue of the above. +type celValue = interface{} + +// ---------------------------------------------------------------------------- +// Tokenizer +// ---------------------------------------------------------------------------- + +type tokKind int + +const ( + tokEOF tokKind = iota + tokIdent + tokInt + tokStr + tokOp // ==, !=, >=, <=, >, <, (, ), . + tokIn // "in" + tokAnd // "and" + tokOr // "or" + tokNot // "not" +) + +type token struct { + kind tokKind + text string +} + +type lexer struct { + src string + pos int +} + +func (l *lexer) next() (token, error) { + for l.pos < len(l.src) && unicode.IsSpace(rune(l.src[l.pos])) { + l.pos++ + } + if l.pos >= len(l.src) { + return token{kind: tokEOF}, nil + } + c := l.src[l.pos] + // string literal + if c == '"' { + start := l.pos + l.pos++ + for l.pos < len(l.src) && l.src[l.pos] != '"' { + l.pos++ + } + if l.pos >= len(l.src) { + return token{}, fmt.Errorf("cel: unterminated string at %d", start) + } + val := l.src[start+1 : l.pos] + l.pos++ // consume closing quote + return token{kind: tokStr, text: val}, nil + } + // integer literal + if unicode.IsDigit(rune(c)) { + start := l.pos + for l.pos < len(l.src) && unicode.IsDigit(rune(l.src[l.pos])) { + l.pos++ + } + return token{kind: tokInt, text: l.src[start:l.pos]}, nil + } + // identifier / keyword + if isIdentStart(c) { + start := l.pos + for l.pos < len(l.src) && isIdentPart(l.src[l.pos]) { + l.pos++ + } + word := l.src[start:l.pos] + switch word { + case "in": + return token{kind: tokIn, text: word}, nil + case "and": + return token{kind: tokAnd, text: word}, nil + case "or": + return token{kind: tokOr, text: word}, nil + case "not": + return token{kind: tokNot, text: word}, nil + default: + return token{kind: tokIdent, text: word}, nil + } + } + // operators + if strings.ContainsRune("()=!<>.", rune(c)) { + // multi-char operators + if l.pos+1 < len(l.src) { + two := l.src[l.pos : l.pos+2] + switch two { + case "==", "!=", ">=", "<=": + l.pos += 2 + return token{kind: tokOp, text: two}, nil + } + } + l.pos++ + return token{kind: tokOp, text: string(c)}, nil + } + return token{}, fmt.Errorf("cel: unexpected character %q at %d", c, l.pos) +} + +func isIdentStart(c byte) bool { + return c == '_' || (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') +} + +func isIdentPart(c byte) bool { + return isIdentStart(c) || (c >= '0' && c <= '9') +} + +// ---------------------------------------------------------------------------- +// Parser (recursive descent, precedence climbing) +// ---------------------------------------------------------------------------- + +type parser struct { + src string + pos int + tok token + err error + node NodeInfo +} + +func newParser(src string, node NodeInfo) *parser { + p := &parser{src: src, node: node} + p.advance() + return p +} + +func (p *parser) len() int { return len(p.src) } + +func (p *parser) advance() { + if p.err != nil { + return + } + l := lexer{src: p.src, pos: p.pos} + t, err := l.next() + if err != nil { + p.err = err + return + } + p.pos = l.pos + p.tok = t +} + +// Grammar (lowest precedence first): +// +// expr := orExpr +// orExpr := andExpr ("or" andExpr)* +// andExpr := notExpr ("and" notExpr)* +// notExpr := "not" notExpr | cmpExpr +// cmpExpr := primary (op primary | "in" primary | "not" "in" primary)? +// primary := "(" expr ")" +// | int +// | str +// | "true" | "false" +// | nodeAttr ("." ident)? // node. +// | ident // bare attribute (e.g. region) +// nodeAttr := "node" + +func (p *parser) parseExpr() (celValue, error) { + if p.err != nil { + return nil, p.err + } + return p.parseOr() +} + +func (p *parser) parseOr() (celValue, error) { + left, err := p.parseAnd() + if err != nil { + return nil, err + } + for p.tok.kind == tokOr { + p.advance() + right, err := p.parseAnd() + if err != nil { + return nil, err + } + lb, ok := left.(bool) + if !ok { + return nil, fmt.Errorf("cel: 'or' operand not bool: %T", left) + } + rb, ok := right.(bool) + if !ok { + return nil, fmt.Errorf("cel: 'or' operand not bool: %T", right) + } + left = lb || rb + } + return left, nil +} + +func (p *parser) parseAnd() (celValue, error) { + left, err := p.parseNot() + if err != nil { + return nil, err + } + for p.tok.kind == tokAnd { + p.advance() + right, err := p.parseNot() + if err != nil { + return nil, err + } + lb, ok := left.(bool) + if !ok { + return nil, fmt.Errorf("cel: 'and' operand not bool: %T", left) + } + rb, ok := right.(bool) + if !ok { + return nil, fmt.Errorf("cel: 'and' operand not bool: %T", right) + } + left = lb && rb + } + return left, nil +} + +func (p *parser) parseNot() (celValue, error) { + if p.tok.kind == tokNot { + // "not" at the start of a primary is logical negation. "not in" + // is handled in parseCmp where it follows a primary. + p.advance() + v, err := p.parseNot() + if err != nil { + return nil, err + } + b, ok := v.(bool) + if !ok { + return nil, fmt.Errorf("cel: 'not' operand not bool: %T", v) + } + return !b, nil + } + return p.parseCmp() +} + +func (p *parser) parseCmp() (celValue, error) { + left, err := p.parsePrimary() + if err != nil { + return nil, err + } + // "not in" + if p.tok.kind == tokNot { + p.advance() + if p.tok.kind != tokIn { + return nil, fmt.Errorf("cel: expected 'in' after 'not', got %q", p.tok.text) + } + p.advance() + right, err := p.parsePrimary() + if err != nil { + return nil, err + } + member, err := inMember(left, right) + if err != nil { + return nil, err + } + return !member, nil + } + // "in" + if p.tok.kind == tokIn { + p.advance() + right, err := p.parsePrimary() + if err != nil { + return nil, err + } + return inMember(left, right) + } + // comparison operators + if p.tok.kind == tokOp { + op := p.tok.text + switch op { + case "==", "!=", ">=", "<=", ">", "<": + p.advance() + right, err := p.parsePrimary() + if err != nil { + return nil, err + } + return compare(op, left, right) + default: + return nil, fmt.Errorf("cel: unexpected operator %q", op) + } + } + return left, nil +} + +// inMember reports whether left is a member of right. right must be a +// list ([]celValue) of comparable values; left may be a string or +// int64. +func inMember(left, right celValue) (bool, error) { + list, ok := right.([]celValue) + if !ok { + return false, fmt.Errorf("cel: 'in' rhs not a list: %T", right) + } + for _, e := range list { + if valuesEqual(left, e) { + return true, nil + } + } + return false, nil +} + +func valuesEqual(a, b celValue) bool { + switch av := a.(type) { + case string: + bv, ok := b.(string) + return ok && av == bv + case int64: + bv, ok := b.(int64) + return ok && av == bv + case bool: + bv, ok := b.(bool) + return ok && av == bv + } + return false +} + +// compare applies a binary comparison operator to two scalar values. +// Strings compare lexicographically; ints numerically; bools only via +// ==/!=. +func compare(op string, left, right celValue) (bool, error) { + switch op { + case "==": + return valuesEqual(left, right), nil + case "!=": + return !valuesEqual(left, right), nil + } + // ordered comparisons require ordered operands + ls, lok := left.(string) + rs, rok := right.(string) + if lok && rok { + switch op { + case "<": + return ls < rs, nil + case "<=": + return ls <= rs, nil + case ">": + return ls > rs, nil + case ">=": + return ls >= rs, nil + } + } + li, lok := left.(int64) + ri, rok := right.(int64) + if lok && rok { + switch op { + case "<": + return li < ri, nil + case "<=": + return li <= ri, nil + case ">": + return li > ri, nil + case ">=": + return li >= ri, nil + } + } + return false, fmt.Errorf("cel: cannot apply %q to %T and %T", op, left, right) +} + +// parsePrimary parses the smallest standalone unit: parenthesised +// expressions, literals, and attribute references. +func (p *parser) parsePrimary() (celValue, error) { + switch p.tok.kind { + case tokOp: + if p.tok.text == "(" { + p.advance() + v, err := p.parseExpr() + if err != nil { + return nil, err + } + if p.tok.kind != tokOp || p.tok.text != ")" { + return nil, fmt.Errorf("cel: expected ')' got %q", p.tok.text) + } + p.advance() + return v, nil + } + return nil, fmt.Errorf("cel: unexpected operator %q", p.tok.text) + case tokInt: + n, err := strconv.ParseInt(p.tok.text, 10, 64) + if err != nil { + return nil, fmt.Errorf("cel: bad int %q: %w", p.tok.text, err) + } + p.advance() + return n, nil + case tokStr: + v := p.tok.text + p.advance() + return v, nil + case tokIdent: + return p.parseAttrRef() + } + return nil, fmt.Errorf("cel: unexpected token %q", p.tok.text) +} + +// parseAttrRef resolves a bare or `node.` attribute reference +// against the node being evaluated. Bare identifiers (e.g. `region`) +// resolve against the same attribute map as `node.region`; the PRD +// examples use both forms interchangeably (see +// TestParseMarkdown_ConstraintsInlineArray). +func (p *parser) parseAttrRef() (celValue, error) { + name := p.tok.text + p.advance() + // dotted access: node. + if p.tok.kind == tokOp && p.tok.text == "." { + if name != "node" { + return nil, fmt.Errorf("cel: dotted access on non-node: %q", name) + } + p.advance() + if p.tok.kind != tokIdent { + return nil, fmt.Errorf("cel: expected attribute name after '.', got %q", p.tok.text) + } + field := p.tok.text + p.advance() + return p.nodeAttr(name + "." + field) + } + // bare identifier + switch name { + case "true": + return true, nil + case "false": + return false, nil + default: + return p.nodeAttr(name) + } +} + +// nodeAttr resolves an attribute name to its value on the parser's +// active node. Mapping (per PRD T2): +// +// node.hostname -> Hostname (string) +// node.kind -> Kind (string) +// node.cpus -> CPU (int64) +// node.memory -> Memory (int64) +// node.tags -> Tags ([]string -> []celValue) +// node.runtimes -> Runtimes ([]string -> []celValue) +// +// Bare names (without the `node.` prefix) resolve through the same +// map, so `region == "us"` and `node.region == "us"` are equivalent +// when the attribute exists. +func (p *parser) nodeAttr(name string) (celValue, error) { + switch name { + case "node.hostname", "hostname": + return p.node.Hostname, nil + case "node.kind", "kind": + return p.node.Kind, nil + case "node.cpus", "cpus": + return p.node.CPU, nil + case "node.memory", "memory": + return p.node.Memory, nil + case "node.tags", "tags": + return toStringValues(p.node.Tags), nil + case "node.runtimes", "runtimes": + return toStringValues(p.node.Runtimes), nil + } + return nil, fmt.Errorf("cel: unknown attribute %q", name) +} + +// toStringValues converts a []string to []celValue so the membership +// operators can compare element-wise. +func toStringValues(in []string) []celValue { + out := make([]celValue, len(in)) + for i, s := range in { + out[i] = s + } + return out +} diff --git a/internal/scheduler/cel_test.go b/internal/scheduler/cel_test.go new file mode 100644 index 0000000..d8b7cdb --- /dev/null +++ b/internal/scheduler/cel_test.go @@ -0,0 +1,210 @@ +package scheduler + +import "testing" + +func TestEvaluateConstraint_Equality(t *testing.T) { + node := NodeInfo{Hostname: "h-1", Kind: "linux", CPU: 4, Memory: 4096, Tags: []string{"web"}, Runtimes: []string{"process"}} + cases := []struct { + name string + expr string + want bool + }{ + {"hostname eq", `node.hostname == "h-1"`, true}, + {"hostname ne", `node.hostname == "h-2"`, false}, + {"kind eq", `node.kind == "linux"`, true}, + {"kind ne", `node.kind == "proxmox"`, false}, + {"cpus eq", `node.cpus == 4`, true}, + {"memory eq", `node.memory == 4096`, true}, + } + for _, c := range cases { + got, err := EvaluateConstraint(c.expr, node) + if err != nil { + t.Errorf("%s: %v", c.name, err) + continue + } + if got != c.want { + t.Errorf("%s: got %v, want %v", c.name, got, c.want) + } + } +} + +func TestEvaluateConstraint_Comparison(t *testing.T) { + node := NodeInfo{Hostname: "h", Kind: "linux", CPU: 4, Memory: 4096} + cases := []struct { + expr string + want bool + }{ + {"node.cpus >= 2", true}, + {"node.cpus >= 4", true}, + {"node.cpus > 4", false}, + {"node.cpus > 2", true}, + {"node.cpus <= 4", true}, + {"node.cpus < 2", false}, + {"node.cpus != 8", true}, + {"node.cpus == 8", false}, + {"node.memory >= 2048", true}, + {"node.memory < 1024", false}, + } + for _, c := range cases { + got, err := EvaluateConstraint(c.expr, node) + if err != nil { + t.Errorf("%q: %v", c.expr, err) + continue + } + if got != c.want { + t.Errorf("%q: got %v, want %v", c.expr, got, c.want) + } + } +} + +func TestEvaluateConstraint_Membership(t *testing.T) { + node := NodeInfo{Tags: []string{"web", "log-shipper"}, Runtimes: []string{"process", "wasmtime"}} + cases := []struct { + expr string + want bool + }{ + {`"web" in node.tags`, true}, + {`"missing" in node.tags`, false}, + {`"process" in node.runtimes`, true}, + {`"podman" in node.runtimes`, false}, + {`"log-shipper" not in node.tags`, false}, + {`"missing" not in node.tags`, true}, + } + for _, c := range cases { + got, err := EvaluateConstraint(c.expr, node) + if err != nil { + t.Errorf("%q: %v", c.expr, err) + continue + } + if got != c.want { + t.Errorf("%q: got %v, want %v", c.expr, got, c.want) + } + } +} + +func TestEvaluateConstraint_BooleanComposition(t *testing.T) { + node := NodeInfo{Kind: "linux", CPU: 4, Tags: []string{"web"}} + cases := []struct { + expr string + want bool + }{ + {`node.kind == "linux" and node.cpus >= 2`, true}, + {`node.kind == "proxmox" and node.cpus >= 2`, false}, + {`node.kind == "linux" or node.kind == "proxmox"`, true}, + {`node.kind == "proxmox" or node.kind == "linux"`, true}, + {`not node.kind == "proxmox"`, true}, + {`not node.kind == "linux"`, false}, + {`(node.kind == "linux") and (node.cpus >= 2)`, true}, + {`node.cpus >= 2 and not "blocked" in node.tags`, true}, + {`node.kind == "linux" and node.cpus >= 2 and "web" in node.tags`, true}, + {`node.kind == "linux" or node.kind == "proxmox" or node.cpus > 100`, true}, + } + for _, c := range cases { + got, err := EvaluateConstraint(c.expr, node) + if err != nil { + t.Errorf("%q: %v", c.expr, err) + continue + } + if got != c.want { + t.Errorf("%q: got %v, want %v", c.expr, got, c.want) + } + } +} + +func TestEvaluateConstraint_BareIdentifiers(t *testing.T) { + // Bare identifiers resolve through the same attribute map as + // node. (per PRD: constraints may use either form). + node := NodeInfo{Kind: "linux", CPU: 4} + got, err := EvaluateConstraint(`kind == "linux"`, node) + if err != nil { + t.Fatalf("bare kind: %v", err) + } + if !got { + t.Error("bare kind == linux: got false, want true") + } +} + +func TestEvaluateConstraint_TrueFalseLiterals(t *testing.T) { + node := NodeInfo{} + cases := []struct { + expr string + want bool + }{ + {"true", true}, + {"false", false}, + {"not false", true}, + {"not true", false}, + {"true and true", true}, + {"true and false", false}, + {"false or true", true}, + } + for _, c := range cases { + got, err := EvaluateConstraint(c.expr, node) + if err != nil { + t.Errorf("%q: %v", c.expr, err) + continue + } + if got != c.want { + t.Errorf("%q: got %v, want %v", c.expr, got, c.want) + } + } +} + +func TestEvaluateConstraint_Errors(t *testing.T) { + node := NodeInfo{Kind: "linux"} + cases := []struct { + name string + expr string + }{ + {"empty", ""}, + {"unterminated string", `node.kind == "linux`}, + {"unknown attribute", `node.bogus == 1`}, + {"unknown bare attr", `bogus == 1`}, + {"dotted on non-node", `host.kind == "linux"`}, + {"bad operator", `node.cpus + 2`}, + {"trailing input", `node.kind == "linux" garbage`}, + {"unbalanced paren", `(node.kind == "linux"`}, + {"missing rhs", `node.cpus >=`}, + {"not without in", `"x" not node.tags`}, + {"ordered compare on bool", `true < false`}, + {"ordered compare on mismatched types", `node.kind > 2`}, + {"in on non-list", `"x" in node.kind`}, + } + for _, c := range cases { + _, err := EvaluateConstraint(c.expr, node) + if err == nil { + t.Errorf("%s: expected error for %q, got nil", c.name, c.expr) + } + } +} + +func TestEvaluateAll(t *testing.T) { + node := NodeInfo{Kind: "linux", CPU: 4, Tags: []string{"web"}} + cases := []struct { + name string + constraints []string + want bool + }{ + {"empty", nil, true}, + {"all pass", []string{`node.kind == "linux"`, "node.cpus >= 2"}, true}, + {"one fails", []string{`node.kind == "linux"`, "node.cpus >= 8"}, false}, + {"all fail", []string{`node.kind == "proxmox"`, "node.cpus >= 8"}, false}, + } + for _, c := range cases { + got, err := EvaluateAll(c.constraints, node) + if err != nil { + t.Errorf("%s: %v", c.name, err) + continue + } + if got != c.want { + t.Errorf("%s: got %v, want %v", c.name, got, c.want) + } + } +} + +func TestEvaluateAll_PropagatesError(t *testing.T) { + node := NodeInfo{} + if _, err := EvaluateAll([]string{"bogus == 1"}, node); err == nil { + t.Error("EvaluateAll: expected error for malformed constraint") + } +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go new file mode 100644 index 0000000..b7cb49f --- /dev/null +++ b/internal/scheduler/scheduler.go @@ -0,0 +1,472 @@ +// Package scheduler implements the v0.9 CLI-side scheduler (REQ-083, +// P05). Unlike the v0.8 daemon-side best-fit scheduler +// (internal/engine/scheduler.go), this scheduler runs entirely in the +// `orca` CLI process (R-001) and is pure: it takes a list of candidate +// nodes plus a workload request and returns placement decisions +// without performing any I/O. +// +// The scheduler is runtime-aware: a workload that declares +// `runtime.one_of: wasm` is only placed on nodes that expose +// `wasmtime` in their Runtimes list; a `pve-vm` workload is only +// placed on `proxmox` nodes. It is also constraint- and +// affinity-aware via the CEL-subset evaluator in cel.go. +// +// Workload kinds are handled differently per the PRD: +// +// - Job: one-shot, returns exactly one placement (best-fit +// bin-packing). +// - Service: count replicas spread across distinct nodes +// (anti-affinity by default); if fewer distinct nodes than count, +// colocation is permitted but distinct nodes are preferred. +// - DaemonSet: one placement per node that fits the constraints. +package scheduler + +import ( + "fmt" + "sort" + + "git.cloudinit.dev/coreci/orca/internal/jobspec" +) + +// NodeInfo is the scheduler's projection of a peer node: total and +// free capacity, the runtimes the node advertises, its tags, and its +// kind (linux/proxmox). The CLI populates this from the +// cluster/peers/ inventory plus the per-node capacity reports +// collected over SSH; the scheduler itself never reads either. +type NodeInfo struct { + Hostname string + Runtimes []string + Tags []string + CPU int64 + Memory int64 + FreeCPU int64 + FreeMem int64 + Kind string +} + +// WorkloadRequest bundles a parsed WorkloadSpec with the namespace +// the workload is being scheduled into. The namespace is carried +// through to placement so the resulting AllocID can be namespaced, +// but the scheduler itself does not inspect it for fitting decisions. +type WorkloadRequest struct { + Spec *jobspec.WorkloadSpec + Namespace string +} + +// Placement is a single scheduling decision: which node, which +// allocation id, and the bin-packing score that won the node the +// placement. AllocID is `ns/spec.Name-` so a multi-replica +// Service produces distinct ids per replica. +type Placement struct { + Node string + AllocID string + Score int64 +} + +// Schedule is the main entry point. For Job (kind=Job) it returns one +// placement on the best-fit node. For Service it returns `Count` +// placements spread across distinct nodes where possible (anti- +// affinity), permitting colocation when Count > nodes. For DaemonSet +// it returns one placement per node that fits. Any kind-agnostic +// validation error (no spec, unknown kind, no fitting node) is +// returned as an error rather than an empty slice so callers can +// distinguish "nothing fits" from "scheduled zero replicas". +func Schedule(nodes []NodeInfo, req WorkloadRequest) ([]Placement, error) { + if req.Spec == nil { + return nil, fmt.Errorf("scheduler: nil WorkloadSpec") + } + if len(nodes) == 0 { + return nil, fmt.Errorf("scheduler: no candidate nodes") + } + switch req.Spec.Kind { + case "Job": + return scheduleJob(nodes, req) + case "Service": + return scheduleService(nodes, req) + case "DaemonSet": + return scheduleDaemonSet(nodes, req) + default: + return nil, fmt.Errorf("scheduler: unknown kind %q", req.Spec.Kind) + } +} + +// Score evaluates a single node against a workload. fits is true iff +// the node (a) advertises a runtime compatible with the workload's +// `runtime.one_of`, (b) satisfies every CEL constraint in +// `spec.Constraints`, and (c) has enough free CPU+memory for the +// workload's requested resources. When fits is true, score is the +// bin-packing score (more free capacity = higher score, so the node +// most likely to absorb the workload without starving its +// neighbours wins). When fits is false, score is 0. +func Score(node NodeInfo, req WorkloadRequest) (score int64, fits bool) { + // (a) runtime compatibility. A workload with no Runtime block or + // an empty OneOf is treated as runtime-agnostic (always fits on + // the runtime axis); this matches the v0.8 behaviour where a + // missing runtime meant "process". + runtimeOK := true + if req.Spec != nil && req.Spec.Runtime != nil && req.Spec.Runtime.OneOf != "" { + runtimeOK = hasRuntime(node, req.Spec.Runtime.OneOf) + } + if !runtimeOK { + return 0, false + } + // (b) constraints. Evaluation errors are treated as non-fit so a + // malformed constraint does not crash Schedule; the caller still + // sees the node filtered out. + if req.Spec != nil { + ok, err := EvaluateAll(req.Spec.Constraints, node) + if err != nil || !ok { + return 0, false + } + } + // (c) capacity. A workload with no Resources block is treated as + // zero-sized for fitting purposes (it always fits the capacity + // axis); real workloads declare cpu/memory. + needCPU, needMem := workloadResources(req) + if node.FreeCPU < needCPU || node.FreeMem < needMem { + return 0, false + } + // bin-packing score: most free capacity wins. CPU is weighted + // 1000x memory so a 1-core difference outweighs a 1-MiB + // difference, mirroring the v0.8 Score weighting that biased + // toward CPU (the more common binding constraint). + score = (node.FreeCPU-needCPU)*1000 + (node.FreeMem - needMem) + if score < 0 { + score = 0 + } + return score, true +} + +// hasRuntime reports whether node advertises the requested runtime. +// The match is case-insensitive and tolerant of aliases: `wasm` and +// `wasmtime` are treated as the same runtime, and `pve-vm`/`pve-ct` +// only match nodes whose Kind is "proxmox". +func hasRuntime(node NodeInfo, oneOf string) bool { + want := normalizeRuntime(oneOf) + // pve-* runtimes require a proxmox-kind node regardless of the + // node's Runtimes list (a proxmox node doesn't list "pve-vm" in + // Runtimes; it IS the runtime). + switch want { + case "pve-vm", "pve-ct", "proxmox": + return normalizeKind(node.Kind) == "proxmox" + } + for _, r := range node.Runtimes { + if normalizeRuntime(r) == want { + return true + } + // alias: wasmtime nodes advertise "wasmtime"; workloads ask + // for "wasm". + if want == "wasm" && normalizeRuntime(r) == "wasmtime" { + return true + } + } + return false +} + +// normalizeRuntime lowercases and trims a runtime name for matching. +func normalizeRuntime(s string) string { + s = toLowerASCII(s) + switch s { + case "wasmtime": + return "wasm" + } + return s +} + +// normalizeKind lowercases and trims a node Kind for matching. +func normalizeKind(s string) string { return toLowerASCII(s) } + +// toLowerASCII lowercases ASCII letters without bringing in strings +// (avoid an alloc-heavy stdlib call in the hot path). +func toLowerASCII(s string) string { + b := []byte(s) + for i, c := range b { + if c >= 'A' && c <= 'Z' { + b[i] = c + 32 + } + } + return string(b) +} + +// workloadResources returns the (cpu, memory) the workload requests, +// read from the WorkloadSpec's Resources block if present. The +// v0.9-P05 WorkloadSpec does not yet carry a Resources field (it +// lands in P0c, REQ-074); until then this returns (0, 0) so the +// capacity check is a no-op and runtime/constraints do the real +// filtering. The signature is here so the scheduler logic does not +// need to change when Resources lands. +func workloadResources(req WorkloadRequest) (int64, int64) { + _ = req + return 0, 0 +} + +// ---------------------------------------------------------------------------- +// Kind-specific scheduling +// ---------------------------------------------------------------------------- + +// scheduleJob places a single Job on the best-fit node. +func scheduleJob(nodes []NodeInfo, req WorkloadRequest) ([]Placement, error) { + type cand struct { + node NodeInfo + score int64 + } + var cands []cand + for _, n := range nodes { + s, ok := Score(n, req) + if !ok { + continue + } + cands = append(cands, cand{node: n, score: s}) + } + // CEL-based affinity rules apply to single-shot Jobs too: a Job + // with `affinity: [{target: "\"ssd\" in node.tags", weight: 100}]` + // should land on the tagged node even without prior placements. + // Name-based affinity (no prior placements to check) contributes + // zero for a standalone Job, so it is harmless to call here. + for i := range cands { + cands[i].score += affinityScore(cands[i].node, req, nil) + } + if len(cands) == 0 { + return nil, fmt.Errorf("scheduler: no node fits workload %q", req.Spec.Name) + } + sort.SliceStable(cands, func(i, j int) bool { + if cands[i].score != cands[j].score { + return cands[i].score > cands[j].score + } + return cands[i].node.Hostname < cands[j].node.Hostname + }) + w := cands[0] + return []Placement{{ + Node: w.node.Hostname, + AllocID: allocID(req, 0), + Score: w.score, + }}, nil +} + +// scheduleService places `Count` replicas with implicit anti-affinity: +// prefer distinct nodes, but permit colocation when Count exceeds the +// number of fitting nodes. Each replica gets a distinct AllocID. +func scheduleService(nodes []NodeInfo, req WorkloadRequest) ([]Placement, error) { + count := req.Spec.Count + if count <= 0 { + count = 1 + } + // Pre-filter fitting nodes once; the loop below re-scores them + // after each placement so the capacity accounting reflects the + // replicas already placed. + fitting := filterFitting(nodes, req) + if len(fitting) == 0 { + return nil, fmt.Errorf("scheduler: no node fits service %q", req.Spec.Name) + } + var placements []Placement + placed := map[string]int{} // hostname -> count placed there + // First pass: spread across distinct nodes. + for i := 0; i < count; i++ { + best, score, ok := pickServiceNode(fitting, req, placements, placed) + if !ok { + break + } + placements = append(placements, Placement{ + Node: best.Hostname, + AllocID: allocID(req, i), + Score: score, + }) + placed[best.Hostname]++ + // Reflect the consumed capacity in the candidate snapshot so + // subsequent picks see updated free capacity. + needCPU, needMem := workloadResources(req) + for j := range fitting { + if fitting[j].Hostname == best.Hostname { + fitting[j].FreeCPU -= needCPU + fitting[j].FreeMem -= needMem + } + } + } + if len(placements) < count { + return nil, fmt.Errorf("scheduler: only placed %d/%d replicas for service %q", + len(placements), count, req.Spec.Name) + } + return placements, nil +} + +// pickServiceNode selects the best node for the next replica. The +// selection prefers nodes with zero prior placements of this service +// (anti-affinity) and applies affinity scoring on top of the +// bin-packing score. +func pickServiceNode(fitting []NodeInfo, req WorkloadRequest, placements []Placement, placed map[string]int) (NodeInfo, int64, bool) { + type scored struct { + node NodeInfo + score int64 + } + var cands []scored + for _, n := range fitting { + s, ok := Score(n, req) + if !ok { + continue + } + // Implicit anti-affinity: a node with N prior replicas of this + // service incurs a penalty of N * (1 << 62) so distinct nodes + // are preferred, but colocation is permitted (with a + // per-replica penalty) when no distinct node remains. This + // produces a balanced spread (e.g. 5 replicas on 3 nodes → + // 2/2/1) rather than stacking everything on the first node. + if placed[n.Hostname] > 0 { + s -= int64(placed[n.Hostname]) * (1 << 62) + } + // Affinity rules from the spec add/subtract their weight. + s += affinityScore(n, req, placements) + cands = append(cands, scored{node: n, score: s}) + } + if len(cands) == 0 { + return NodeInfo{}, 0, false + } + sort.SliceStable(cands, func(i, j int) bool { + if cands[i].score != cands[j].score { + return cands[i].score > cands[j].score + } + return cands[i].node.Hostname < cands[j].node.Hostname + }) + w := cands[0] + return w.node, w.score, true +} + +// scheduleDaemonSet places one replica per node that fits the +// constraints. The PRD's DaemonSet placement mode (every-node / +// matching / mandatory) lives on the WorkloadSpec.Schedule block; the +// scheduler honours it indirectly by filtering on Constraints: a +// `matching` DaemonSet carries constraints that select the matching +// nodes, an `every-node` DaemonSet carries none, and a `mandatory` +// one is enforced elsewhere (the scheduler still just returns +// placements for every fitting node). +func scheduleDaemonSet(nodes []NodeInfo, req WorkloadRequest) ([]Placement, error) { + var placements []Placement + for _, n := range nodes { + s, ok := Score(n, req) + if !ok { + continue + } + placements = append(placements, Placement{ + Node: n.Hostname, + AllocID: allocID(req, len(placements)), + Score: s, + }) + } + if len(placements) == 0 { + return nil, fmt.Errorf("scheduler: no node fits daemonset %q", req.Spec.Name) + } + return placements, nil +} + +// ---------------------------------------------------------------------------- +// Affinity scoring +// ---------------------------------------------------------------------------- + +// affinityScore returns the weighted affinity contribution for a +// node given the placements already made. For each AffinityRule the +// Target is a CEL expression; if it evaluates true against the node, +// the rule's Weight is added (positive = co-locate, negative = +// anti-affinity). An affinity target that fails to evaluate is +// ignored rather than failing the schedule: operators use affinity as +// a hint, not a hard gate. +// +// The PRD also mentions affinity rules like `{target: "redis", weight: +// 50}` where Target is a workload *name* rather than a CEL expression. +// We support both: if Target parses as a CEL expression it is +// evaluated against the node; otherwise it is treated as a workload +// name and we check whether any already-placed alloc for that name +// exists on the node. The placement-already-here check is done by the +// caller via placements; this function checks the node's own +// attributes only. +func affinityScore(node NodeInfo, req WorkloadRequest, placements []Placement) int64 { + if req.Spec == nil { + return 0 + } + var total int64 + for _, rule := range req.Spec.Affinity { + // Try CEL evaluation first; if the target is a bare workload + // name (no operator) the CEL parser will fail and we fall + // back to name-based placement counting. + ok, err := EvaluateConstraint(rule.Target, node) + if err == nil { + if ok { + total += int64(rule.Weight) + } + continue + } + // Fallback: target is a workload name; count existing + // placements for that workload on this node and apply the + // weight once per co-located replica. + for _, p := range placements { + if p.Node == node.Hostname && isAllocFor(p.AllocID, rule.Target) { + total += int64(rule.Weight) + } + } + } + return total +} + +// isAllocFor reports whether an AllocID encodes a placement for the +// named workload. AllocIDs are `ns/name-idx`, so we look for the +// workload name as the segment after the first slash and before the +// trailing `-idx`. +func isAllocFor(allocID, workloadName string) bool { + // strip namespace prefix + rest := allocID + if i := indexByte(rest, '/'); i >= 0 { + rest = rest[i+1:] + } + // strip trailing -idx + if i := lastIndexByte(rest, '-'); i >= 0 { + rest = rest[:i] + } + return rest == workloadName +} + +// indexByte returns the index of the first occurrence of b in s, or +// -1. Avoids importing strings just for one helper. +func indexByte(s string, b byte) int { + for i := 0; i < len(s); i++ { + if s[i] == b { + return i + } + } + return -1 +} + +// lastIndexByte returns the index of the last occurrence of b in s, or +// -1. +func lastIndexByte(s string, b byte) int { + for i := len(s) - 1; i >= 0; i-- { + if s[i] == b { + return i + } + } + return -1 +} + +// ---------------------------------------------------------------------------- +// Helpers +// ---------------------------------------------------------------------------- + +// filterFitting returns a copy of the nodes that pass Score for the +// request, preserving order. Capacity is not yet decremented; the +// caller adjusts FreeCPU/FreeMem as it places replicas. +func filterFitting(nodes []NodeInfo, req WorkloadRequest) []NodeInfo { + var out []NodeInfo + for _, n := range nodes { + if _, ok := Score(n, req); ok { + out = append(out, n) + } + } + return out +} + +// allocID renders a stable, namespaced allocation id for a placement. +// Format: `ns/spec.Name-`. +func allocID(req WorkloadRequest, idx int) string { + ns := req.Namespace + if ns == "" { + ns = "default" + } + return fmt.Sprintf("%s/%s-%d", ns, req.Spec.Name, idx) +} diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go new file mode 100644 index 0000000..b68cbd0 --- /dev/null +++ b/internal/scheduler/scheduler_test.go @@ -0,0 +1,451 @@ +package scheduler + +import ( + "strings" + "testing" + + "git.cloudinit.dev/coreci/orca/internal/jobspec" +) + +// threeLinuxNodes returns a small cluster of three Linux nodes with +// distinct free capacities so best-fit ordering is unambiguous. +func threeLinuxNodes() []NodeInfo { + return []NodeInfo{ + {Hostname: "node-a", Runtimes: []string{"process"}, Tags: nil, CPU: 4, Memory: 4096, FreeCPU: 4, FreeMem: 4096, Kind: "linux"}, + {Hostname: "node-b", Runtimes: []string{"process"}, Tags: nil, CPU: 8, Memory: 8192, FreeCPU: 8, FreeMem: 8192, Kind: "linux"}, + {Hostname: "node-c", Runtimes: []string{"process"}, Tags: nil, CPU: 2, Memory: 2048, FreeCPU: 2, FreeMem: 2048, Kind: "linux"}, + } +} + +func jobSpec(name, oneOf string, constraints []string) *jobspec.WorkloadSpec { + return &jobspec.WorkloadSpec{ + Kind: "Job", + Name: name, + Count: 1, + Runtime: &jobspec.RuntimeBlock{OneOf: oneOf}, + Constraints: constraints, + } +} + +func serviceSpec(name, oneOf string, count int, constraints []string) *jobspec.WorkloadSpec { + return &jobspec.WorkloadSpec{ + Kind: "Service", + Name: name, + Count: count, + Runtime: &jobspec.RuntimeBlock{OneOf: oneOf}, + Constraints: constraints, + } +} + +func daemonSetSpec(name, oneOf string, constraints []string) *jobspec.WorkloadSpec { + return &jobspec.WorkloadSpec{ + Kind: "DaemonSet", + Name: name, + Count: 1, + Runtime: &jobspec.RuntimeBlock{OneOf: oneOf}, + Constraints: constraints, + } +} + +// --------------------------------------------------------------------------- +// Job +// --------------------------------------------------------------------------- + +func TestScheduleJob_BestFit(t *testing.T) { + nodes := threeLinuxNodes() + req := WorkloadRequest{Spec: jobSpec("batch", "process", nil), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if len(got) != 1 { + t.Fatalf("placements = %d, want 1", len(got)) + } + if got[0].Node != "node-b" { + t.Errorf("Node = %q, want node-b (most free capacity)", got[0].Node) + } + if !strings.HasPrefix(got[0].AllocID, "ns/batch-") { + t.Errorf("AllocID = %q, want ns/batch-*", got[0].AllocID) + } + if got[0].Score <= 0 { + t.Errorf("Score = %d, want > 0", got[0].Score) + } +} + +func TestScheduleJob_NoFittingNode(t *testing.T) { + nodes := threeLinuxNodes() + // wasm runtime not advertised by any node. + req := WorkloadRequest{Spec: jobSpec("wasmjob", "wasm", nil), Namespace: "ns"} + if _, err := Schedule(nodes, req); err == nil { + t.Fatal("Schedule: expected error for no-fitting node, got nil") + } +} + +// --------------------------------------------------------------------------- +// Service +// --------------------------------------------------------------------------- + +func TestScheduleService_SpreadAcrossNodes(t *testing.T) { + nodes := threeLinuxNodes() + req := WorkloadRequest{Spec: serviceSpec("web", "process", 3, nil), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if len(got) != 3 { + t.Fatalf("placements = %d, want 3", len(got)) + } + seen := map[string]int{} + for _, p := range got { + seen[p.Node]++ + } + if len(seen) != 3 { + t.Errorf("anti-affinity spread: distinct nodes = %d, want 3; %v", len(seen), seen) + } +} + +func TestScheduleService_ColocationWhenFewerNodes(t *testing.T) { + nodes := threeLinuxNodes() + req := WorkloadRequest{Spec: serviceSpec("web", "process", 5, nil), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if len(got) != 5 { + t.Fatalf("placements = %d, want 5", len(got)) + } + seen := map[string]int{} + for _, p := range got { + seen[p.Node]++ + } + if len(seen) != 3 { + t.Errorf("colocation: distinct nodes = %d, want 3 (all used)", len(seen)) + } + // No node should host more than 2 (3 nodes, 5 replicas: 2+2+1). + for n, c := range seen { + if c > 2 { + t.Errorf("node %s has %d replicas, want <= 2", n, c) + } + } +} + +func TestScheduleService_NoFittingNode(t *testing.T) { + nodes := threeLinuxNodes() + req := WorkloadRequest{Spec: serviceSpec("wasm-svc", "wasm", 3, nil), Namespace: "ns"} + if _, err := Schedule(nodes, req); err == nil { + t.Fatal("Schedule: expected error for service with no fitting node") + } +} + +// --------------------------------------------------------------------------- +// DaemonSet +// --------------------------------------------------------------------------- + +func TestScheduleDaemonSet_AllMatching(t *testing.T) { + nodes := threeLinuxNodes() + req := WorkloadRequest{Spec: daemonSetSpec("logrotate", "process", nil), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if len(got) != 3 { + t.Errorf("placements = %d, want 3 (one per node)", len(got)) + } + seen := map[string]bool{} + for _, p := range got { + seen[p.Node] = true + } + if len(seen) != 3 { + t.Errorf("DaemonSet distinct nodes = %d, want 3", len(seen)) + } +} + +func TestScheduleDaemonSet_SomeExcludedByConstraint(t *testing.T) { + nodes := threeLinuxNodes() + // Only nodes with cpus >= 4 qualify: node-a (4) and node-b (8). + req := WorkloadRequest{Spec: daemonSetSpec("heavy", "process", []string{"node.cpus >= 4"}), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if len(got) != 2 { + t.Errorf("placements = %d, want 2 (cpus>=4)", len(got)) + } +} + +// --------------------------------------------------------------------------- +// Runtime compatibility +// --------------------------------------------------------------------------- + +func TestSchedule_RuntimeCompatibilityWasm(t *testing.T) { + nodes := []NodeInfo{ + {Hostname: "no-wasm", Runtimes: []string{"process"}, Kind: "linux", CPU: 8, Memory: 8192, FreeCPU: 8, FreeMem: 8192}, + {Hostname: "has-wasm", Runtimes: []string{"process", "wasmtime"}, Kind: "linux", CPU: 4, Memory: 4096, FreeCPU: 4, FreeMem: 4096}, + } + // Even though no-wasm has more free capacity, the wasm workload + // must land on has-wasm. + req := WorkloadRequest{Spec: jobSpec("wasmjob", "wasm", nil), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if got[0].Node != "has-wasm" { + t.Errorf("Node = %q, want has-wasm (runtime compatibility)", got[0].Node) + } +} + +func TestSchedule_RuntimeCompatibilityPveVM(t *testing.T) { + nodes := []NodeInfo{ + {Hostname: "linux-1", Runtimes: []string{"process"}, Kind: "linux", CPU: 8, Memory: 8192, FreeCPU: 8, FreeMem: 8192}, + {Hostname: "pve-1", Runtimes: []string{"process"}, Kind: "proxmox", CPU: 8, Memory: 8192, FreeCPU: 8, FreeMem: 8192}, + } + req := WorkloadRequest{Spec: jobSpec("vmjob", "pve-vm", nil), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if got[0].Node != "pve-1" { + t.Errorf("Node = %q, want pve-1 (pve-vm requires proxmox kind)", got[0].Node) + } +} + +// --------------------------------------------------------------------------- +// Constraints +// --------------------------------------------------------------------------- + +func TestSchedule_ConstraintKindExcludesProxmox(t *testing.T) { + nodes := []NodeInfo{ + {Hostname: "linux-1", Runtimes: []string{"process"}, Kind: "linux", CPU: 8, Memory: 8192, FreeCPU: 8, FreeMem: 8192}, + {Hostname: "pve-1", Runtimes: []string{"process"}, Kind: "proxmox", CPU: 8, Memory: 8192, FreeCPU: 8, FreeMem: 8192}, + } + req := WorkloadRequest{Spec: jobSpec("linuxonly", "process", []string{`node.kind == "linux"`}), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if got[0].Node != "linux-1" { + t.Errorf("Node = %q, want linux-1 (kind==linux)", got[0].Node) + } +} + +func TestSchedule_ConstraintCPUsExcludesSmall(t *testing.T) { + nodes := threeLinuxNodes() // node-c has cpus=2 + req := WorkloadRequest{Spec: jobSpec("big", "process", []string{"node.cpus >= 4"}), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if got[0].Node == "node-c" { + t.Errorf("Node = node-c, want node-a or node-b (cpus>=4)") + } +} + +func TestSchedule_ConstraintNotInTags(t *testing.T) { + nodes := []NodeInfo{ + {Hostname: "tagged", Runtimes: []string{"process"}, Tags: []string{"log-shipper"}, Kind: "linux", CPU: 8, Memory: 8192, FreeCPU: 8, FreeMem: 8192}, + {Hostname: "clean", Runtimes: []string{"process"}, Tags: nil, Kind: "linux", CPU: 4, Memory: 4096, FreeCPU: 4, FreeMem: 4096}, + } + req := WorkloadRequest{Spec: jobSpec("worker", "process", []string{`"log-shipper" not in node.tags`}), Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if got[0].Node != "clean" { + t.Errorf("Node = %q, want clean (log-shipper not in tags)", got[0].Node) + } +} + +// --------------------------------------------------------------------------- +// Affinity +// --------------------------------------------------------------------------- + +func TestSchedule_AffinityPrefersColocatedNode(t *testing.T) { + // Place a redis service first, then a worker with affinity for + // redis; the worker should prefer the node where redis already + // runs even if another node has more free capacity. + nodes := []NodeInfo{ + {Hostname: "big", Runtimes: []string{"process"}, Kind: "linux", CPU: 8, Memory: 8192, FreeCPU: 8, FreeMem: 8192}, + {Hostname: "small", Runtimes: []string{"process"}, Kind: "linux", CPU: 4, Memory: 4096, FreeCPU: 4, FreeMem: 4096}, + } + redisReq := WorkloadRequest{Spec: serviceSpec("redis", "process", 1, nil), Namespace: "ns"} + redisPlacements, err := Schedule(nodes, redisReq) + if err != nil { + t.Fatalf("redis Schedule: %v", err) + } + // Redis lands on "big" (most free capacity). Now schedule the + // worker with affinity to redis; it should also land on "big". + workerReq := WorkloadRequest{Spec: &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "worker", + Count: 1, + Runtime: &jobspec.RuntimeBlock{OneOf: "process"}, + Affinity: []jobspec.AffinityRule{ + {Target: "redis", Weight: 1000}, + }, + }, Namespace: "ns"} + // The affinity is name-based; we need to seed the worker schedule + // with the redis placement so affinityScore can see it. Schedule + // does not take prior placements, so test affinityScore directly. + got := affinityScore(nodes[0], workerReq, redisPlacements) + if got <= 0 { + t.Errorf("affinityScore(big) = %d, want > 0 (redis colocated)", got) + } + gotSmall := affinityScore(nodes[1], workerReq, redisPlacements) + if gotSmall != 0 { + t.Errorf("affinityScore(small) = %d, want 0 (redis not colocated)", gotSmall) + } +} + +func TestSchedule_AffinityCELExpression(t *testing.T) { + // Affinity with a CEL target: prefer nodes tagged "ssd". + nodes := []NodeInfo{ + {Hostname: "hdd", Runtimes: []string{"process"}, Tags: []string{"hdd"}, Kind: "linux", CPU: 8, Memory: 8192, FreeCPU: 8, FreeMem: 8192}, + {Hostname: "ssd", Runtimes: []string{"process"}, Tags: []string{"ssd"}, Kind: "linux", CPU: 4, Memory: 4096, FreeCPU: 4, FreeMem: 4096}, + } + req := WorkloadRequest{Spec: &jobspec.WorkloadSpec{ + Kind: "Job", + Name: "db", + Count: 1, + Runtime: &jobspec.RuntimeBlock{OneOf: "process"}, + Affinity: []jobspec.AffinityRule{ + {Target: `"ssd" in node.tags`, Weight: 10000}, + }, + }, Namespace: "ns"} + got, err := Schedule(nodes, req) + if err != nil { + t.Fatalf("Schedule: %v", err) + } + if got[0].Node != "ssd" { + t.Errorf("Node = %q, want ssd (affinity to ssd tag outweighs capacity)", got[0].Node) + } +} + +// --------------------------------------------------------------------------- +// Error paths +// --------------------------------------------------------------------------- + +func TestSchedule_EmptyNodes(t *testing.T) { + req := WorkloadRequest{Spec: jobSpec("x", "process", nil), Namespace: "ns"} + if _, err := Schedule(nil, req); err == nil { + t.Fatal("Schedule: expected error for empty nodes, got nil") + } +} + +func TestSchedule_NilSpec(t *testing.T) { + if _, err := Schedule(threeLinuxNodes(), WorkloadRequest{}); err == nil { + t.Fatal("Schedule: expected error for nil spec, got nil") + } +} + +func TestSchedule_UnknownKind(t *testing.T) { + req := WorkloadRequest{Spec: &jobspec.WorkloadSpec{Kind: "Cron", Name: "x", Count: 1}, Namespace: "ns"} + if _, err := Schedule(threeLinuxNodes(), req); err == nil { + t.Fatal("Schedule: expected error for unknown kind") + } +} + +// --------------------------------------------------------------------------- +// Score unit tests +// --------------------------------------------------------------------------- + +func TestScore_FitsAndDoesNotFit(t *testing.T) { + node := NodeInfo{Hostname: "n", Runtimes: []string{"process"}, Kind: "linux", CPU: 4, Memory: 4096, FreeCPU: 4, FreeMem: 4096} + req := WorkloadRequest{Spec: jobSpec("j", "process", nil), Namespace: "ns"} + score, fits := Score(node, req) + if !fits { + t.Error("fits = false, want true") + } + if score <= 0 { + t.Errorf("score = %d, want > 0", score) + } +} + +func TestScore_RuntimeMismatchDoesNotFit(t *testing.T) { + node := NodeInfo{Hostname: "n", Runtimes: []string{"process"}, Kind: "linux", CPU: 4, Memory: 4096, FreeCPU: 4, FreeMem: 4096} + req := WorkloadRequest{Spec: jobSpec("j", "wasm", nil), Namespace: "ns"} + if _, fits := Score(node, req); fits { + t.Error("fits = true for wasm on process-only node, want false") + } +} + +func TestScore_ConstraintFailsDoesNotFit(t *testing.T) { + node := NodeInfo{Hostname: "n", Runtimes: []string{"process"}, Kind: "linux", CPU: 4, Memory: 4096, FreeCPU: 4, FreeMem: 4096} + req := WorkloadRequest{Spec: jobSpec("j", "process", []string{`node.kind == "proxmox"`}), Namespace: "ns"} + if _, fits := Score(node, req); fits { + t.Error("fits = true for kind==proxmox on linux node, want false") + } +} + +// --------------------------------------------------------------------------- +// allocID / isAllocFor helpers +// --------------------------------------------------------------------------- + +func TestAllocID(t *testing.T) { + req := WorkloadRequest{Spec: &jobspec.WorkloadSpec{Name: "web"}, Namespace: "prod"} + if got := allocID(req, 2); got != "prod/web-2" { + t.Errorf("allocID = %q, want prod/web-2", got) + } + req.Namespace = "" + if got := allocID(req, 0); got != "default/web-0" { + t.Errorf("allocID = %q, want default/web-0", got) + } +} + +func TestIsAllocFor(t *testing.T) { + cases := []struct { + allocID string + workload string + want bool + }{ + {"ns/redis-0", "redis", true}, + {"ns/redis-12", "redis", true}, + {"ns/worker-0", "redis", false}, + {"redis-0", "redis", true}, + {"ns/web-canary-3", "web-canary", true}, + } + for _, c := range cases { + if got := isAllocFor(c.allocID, c.workload); got != c.want { + t.Errorf("isAllocFor(%q,%q) = %v, want %v", c.allocID, c.workload, got, c.want) + } + } +} + +// --------------------------------------------------------------------------- +// normalizeRuntime / hasRuntime +// --------------------------------------------------------------------------- + +func TestHasRuntimeAliases(t *testing.T) { + cases := []struct { + name string + node NodeInfo + want bool + }{ + {"wasm on wasmtime node", NodeInfo{Runtimes: []string{"wasmtime"}, Kind: "linux"}, true}, + {"wasm on process node", NodeInfo{Runtimes: []string{"process"}, Kind: "linux"}, false}, + {"pve-vm on linux node", NodeInfo{Runtimes: []string{"pve-vm"}, Kind: "linux"}, false}, + {"pve-vm on proxmox node", NodeInfo{Runtimes: nil, Kind: "proxmox"}, true}, + {"process on process node", NodeInfo{Runtimes: []string{"process"}, Kind: "linux"}, true}, + {"empty runtime on any node", NodeInfo{Runtimes: []string{"process"}, Kind: "linux"}, true}, + } + for _, c := range cases { + if c.name == "empty runtime on any node" { + // hasRuntime is only called when OneOf != "". + continue + } + if got := hasRuntime(c.node, "wasm"); c.name == "wasm on wasmtime node" || c.name == "wasm on process node" { + if got != c.want { + t.Errorf("%s: hasRuntime(wasm) = %v, want %v", c.name, got, c.want) + } + } + } + // Explicit pve-vm and process checks. + if !hasRuntime(NodeInfo{Runtimes: nil, Kind: "proxmox"}, "pve-vm") { + t.Error("pve-vm on proxmox node should fit") + } + if hasRuntime(NodeInfo{Runtimes: nil, Kind: "linux"}, "pve-vm") { + t.Error("pve-vm on linux node should not fit") + } + if !hasRuntime(NodeInfo{Runtimes: []string{"process"}, Kind: "linux"}, "process") { + t.Error("process on process node should fit") + } +}