Compare commits

...

4 Commits

Author SHA1 Message Date
Jon Chery 790109ea24 feat(D): capacity auto-discovery + partial updates + ACL debug + UX (REQ-168,169)
- node capacity set: partial updates (only set dimensions passed;
  read-modify-write on existing row)
- node capacity auto [percentage]: SSH to node, discover CPU (nproc),
  memory (/proc/meminfo), disk (df), multiply by percentage (default 75)
- ACL check --verbose: prints resolved ACLPath + all entries + identity
- job list UX: short 8-char IDs, NODE column, conditional EXIT (- for
  non-terminal statuses)

---ci---
project: orca
milestone: v0.12.18
phase: D
status: complete
requirements:
  covered: [168, 169]
---/ci---
2026-08-10 16:33:27 +00:00
Jon Chery 64cbbd543e feat(C): remote deployment correctness — PVE runtime, DB record, Traefik (REQ-166)
- deployRemote branches on runtime: pve-ct/pve-vm on proxmox nodes
  invoke runtime.Registry.Prepare+Start (creates LXC/VM via SSH);
  process runtime on linux nodes uses systemd emitter; process on
  proxmox is rejected with clear error
- deployRemote emits Traefik dynamic config when spec has ports
  (TraefikEmitter.Render + SSH-push to /etc/traefik/dynamic/)
- model.Job: added Node field so job list --json reports deployed node
- job run remote case: inserts model.Job + alloc_history after
  deployRemote succeeds (job list and job stop now work for remote)
- --target name lookup: legacy dispatcher tries NodeID match
- job list UX: short 8-char IDs, NODE column, conditional EXIT (- for
  non-terminal statuses)
- emitter/traefik.go: directory provider (was single file), pve-ct/pve-vm
  registered in RegisterTraefik
- runtime/pve.go: shellQuote for image, idempotent create check

---ci---
project: orca
milestone: v0.12.18
phase: C
status: complete
requirements:
  covered: [166]
---/ci---
2026-08-10 16:21:50 +00:00
Jon Chery 16440a89f2 feat(B): Traefik deployment to all nodes during init/join (REQ-165, REQ-167)
- internal/traefik/install.go: shared Traefik installer (download +
  systemd unit + dynamic dir). Default v3.3.0, configurable.
- orca init: installs Traefik on localhost (idempotent, non-fatal
  if offline)
- proxmox bootstrap: installs Traefik on PVE host + downloads LXC
  template (default ubuntu-24.04, --lxc-template flag)
- linux bootstrap: installs Traefik on worker
- emitter/traefik.go: directory provider (was single file);
  register pve-ct/pve-vm in RegisterTraefik
- --lxc-template flag on node join (default ubuntu-24.04)

---ci---
project: orca
milestone: v0.12.18
phase: B
status: complete
requirements:
  covered: [165, 167]
---/ci---
2026-08-10 16:09:48 +00:00
Jon Chery a6ceb13491 fix(A): bootstrap plumbing — init creates SSH key + known_hosts + master key (REQ-164)
Fixes UAT issues 1, 8, 9, 12C, 13:
- orca init: generates SSH keypair (GenerateOrLoadSSHKey), creates
  empty known_hosts (0600), generates master key (GenerateMasterKey +
  SaveMasterKey). All were missing from runInit — every downstream
  SSH/secrets/cluster operation failed on a fresh init.
- TOFUHostKeyCallbackPath: creates known_hosts file if it doesn't exist
  (defense-in-depth alongside init)
- Linux bootstrap: replaces buggy inline TOFU with
  proxmox.TOFUHostKeyCallbackPath (first-connect key capture works)
- --type flag help: includes "linux" (was "localhost or proxmox")
- doctor network: SSH exec probe (was HTTP /healthz to :8443 — no
  daemon in SSH-push model R-001)

---ci---
project: orca
milestone: v0.12.18
phase: A
status: complete
requirements:
  covered: [164]
---/ci---
2026-08-10 16:02:19 +00:00
18 changed files with 560 additions and 70 deletions
+12
View File
@@ -32,6 +32,7 @@ var (
aclRevokeNamespace string
aclCheckNamespace string
aclCheckPermission string
aclCheckVerbose bool
)
var aclCmd = &cobra.Command{
@@ -351,6 +352,16 @@ read, write, admin (default: read).`,
}
identity.Namespace = ns
allowed := a.Check(identity, ns, perm)
if aclCheckVerbose {
fmt.Fprintf(cmd.ErrOrStderr(), "ACL path: %s\n", paths.ACLPath())
fmt.Fprintf(cmd.ErrOrStderr(), "Identity: kind=%s id=%s ns=%s\n", identity.Kind, identity.ID, ns)
fmt.Fprintf(cmd.ErrOrStderr(), "Permission: %s -> allowed=%v\n", permStr, allowed)
entries := a.List()
fmt.Fprintf(cmd.ErrOrStderr(), "ACL entries (%d):\n", len(entries))
for _, e := range entries {
fmt.Fprintf(cmd.ErrOrStderr(), " kind=%s id=%s ns=%s perms=%d\n", e.Identity.Kind, e.Identity.ID, e.Namespace, e.Permissions)
}
}
if jsonOutput {
return printJSON(map[string]any{
"identity": identity,
@@ -373,6 +384,7 @@ func init() {
aclGrantCmd.Flags().StringVar(&aclGrantPermissions, "permissions", "read", "comma-separated permissions: read,write,admin")
aclRevokeCmd.Flags().StringVar(&aclRevokeNamespace, "namespace", "", "namespace scope (required for tokens; defaults to spiffe path ns)")
aclCheckCmd.Flags().StringVar(&aclCheckNamespace, "namespace", "", "namespace scope (required for tokens; defaults to spiffe path ns)")
aclCheckCmd.Flags().BoolVar(&aclCheckVerbose, "verbose", false, "print ACL path + loaded entries for debugging")
aclCheckCmd.Flags().StringVar(&aclCheckPermission, "permission", "read", "permission to check: read, write, or admin")
aclCmd.AddCommand(aclGrantCmd)
+85 -1
View File
@@ -6,9 +6,12 @@ import (
"encoding/pem"
"fmt"
"os"
"path/filepath"
"time"
"github.com/google/uuid"
"golang.org/x/crypto/ssh"
"github.com/spf13/cobra"
"git.cloudinit.dev/coreci/orca/internal/acl"
@@ -16,6 +19,7 @@ import (
"git.cloudinit.dev/coreci/orca/internal/identity"
"git.cloudinit.dev/coreci/orca/internal/model"
"git.cloudinit.dev/coreci/orca/internal/paths"
"git.cloudinit.dev/coreci/orca/internal/secrets"
"git.cloudinit.dev/coreci/orca/internal/security"
"git.cloudinit.dev/coreci/orca/internal/store"
)
@@ -59,7 +63,8 @@ func runInit(out interface{ Write([]byte) (int, error) }) error {
Database string `json:"database"`
CAFingerprint string `json:"ca_fingerprint,omitempty"`
CertFingerprint string `json:"cert_fingerprint,omitempty"`
OS string `json:"os"`
OS string `json:"os"
"path/filepath"`
NodeID string `json:"node_id"`
NodeName string `json:"node_name"`
Steps []stepResult `json:"steps"`
@@ -138,6 +143,85 @@ func runInit(out interface{ Write([]byte) (int, error) }) error {
}
}
// Step 4a: SSH keypair (idempotent — GenerateOrLoadSSHKey has a fast-path).
// REQ-164: without this, every sshpush.Transport dial fails because
// the orca SSH key doesn't exist after a fresh init.
sshKeyPEM, sshPubLine, err := security.GenerateOrLoadSSHKey(dir)
if err != nil {
return fmt.Errorf("generate SSH keypair: %w", err)
}
_ = sshKeyPEM
sshKeyFp := ""
if pubKey, err := ssh.ParsePublicKey(sshPubLine); err == nil {
sshKeyFp = ssh.FingerprintSHA256(pubKey)
}
summary.Steps = append(summary.Steps, stepResult{Label: "ssh-key", Status: "ok", Detail: sshKeyFp[:min(16, len(sshKeyFp))] + "..."})
if !jsonOutput {
fmt.Fprintf(out, "\xe2\x9c\x93 SSH keypair provisioned: fp=%s\n", sshKeyFp[:min(16, len(sshKeyFp))]+"...")
}
// Step 4b: known_hosts file (empty, 0600). Without this, the TOFU
// host-key callback fails with "no such file" on the first SSH dial
// (knownhosts.New requires the file to exist).
knownHostsPath := certpaths.KnownHostsPath()
if _, err := os.Stat(knownHostsPath); err != nil {
if os.IsNotExist(err) {
if err := os.WriteFile(knownHostsPath, []byte{}, 0o600); err != nil {
return fmt.Errorf("create known_hosts: %w", err)
}
} else {
return fmt.Errorf("stat known_hosts: %w", err)
}
}
summary.Steps = append(summary.Steps, stepResult{Label: "known-hosts", Status: "ok", Detail: knownHostsPath})
if !jsonOutput {
fmt.Fprintf(out, "\xe2\x9c\x93 Known hosts file created: %s\n", knownHostsPath)
}
// Step 4c: master key (32-byte random, 0600). Without this, secrets
// set/get/rotate and cluster seal/unseal all fail with "stat master
// key: no such file or directory" on a fresh init.
masterKeyPath := paths.MasterKeyPath()
if _, err := os.Stat(masterKeyPath); err != nil {
if os.IsNotExist(err) {
os.MkdirAll(filepath.Dir(masterKeyPath), 0o755)
masterKey, err := secrets.GenerateMasterKey()
if err != nil {
return fmt.Errorf("generate master key: %w", err)
}
if err := secrets.SaveMasterKey(masterKeyPath, masterKey); err != nil {
return fmt.Errorf("save master key: %w", err)
}
// Zero the key from memory (defense-in-depth, REQ-154).
defer secrets.ZeroKey(masterKey)
summary.Steps = append(summary.Steps, stepResult{Label: "master-key", Status: "ok", Detail: "generated"})
if !jsonOutput {
fmt.Fprintf(out, "\xe2\x9c\x93 Master key generated: %s\n", masterKeyPath)
}
} else {
return fmt.Errorf("stat master key: %w", err)
}
} else {
summary.Steps = append(summary.Steps, stepResult{Label: "master-key", Status: "skipped", Detail: "already present"})
if !jsonOutput {
fmt.Fprintf(out, "\xe2\x9c\x93 Master key: already present\n")
}
}
// Step 4d: Install Traefik on the lead node (REQ-165, Phase B).
// Traefik is the data-plane ingress. Idempotent.
if err := installTraefikLocal(); err != nil {
if !jsonOutput {
fmt.Fprintf(out, "Traefik install skipped: %v\n", err)
}
summary.Steps = append(summary.Steps, stepResult{Label: "traefik", Status: "skipped", Detail: err.Error()})
} else {
summary.Steps = append(summary.Steps, stepResult{Label: "traefik", Status: "ok", Detail: traefikVersion})
if !jsonOutput {
fmt.Fprintf(out, "Traefik installed: %s\n", traefikVersion)
}
}
// Step 5: OS detection.
osDetected := detectOS()
summary.OS = osDetected
+47 -7
View File
@@ -115,7 +115,7 @@ var jobRunCmd = &cobra.Command{
switch res.mode {
case "remote":
// Scheduler selected a node (or --target pinned one): render
// the systemd unit, verify it, and SSH-push to the peer.
// the systemd unit / PVE container, verify it, and SSH-push.
// C-44: a push failure is an error (no local fallback).
unitPaths, derr := deployRemote(ctx, spec, res, nodesByHost)
logDispatch(res, derr)
@@ -126,8 +126,12 @@ var jobRunCmd = &cobra.Command{
return derr
}
res.unitPaths = unitPaths
// REQ-156 / P07 T5: invalidate the jobs cache (the
// dispatch decision records a local job entry).
// REQ-166 / Phase C2: insert a Job DB record so `job list`
// and `job stop` can find the remotely-deployed job.
if dbErr := insertRemoteJob(spec, res.node); dbErr != nil {
// Non-fatal: the job is deployed, just not visible to list.
logDispatch(res, dbErr)
}
cacheInvalidate(cacheJobClass)
if jsonOutput {
return printJSON(map[string]any{
@@ -219,9 +223,17 @@ func renderJobs(cmd *cobra.Command, jobs []*model.Job) error {
fmt.Fprintln(cmd.OutOrStdout(), "No jobs. Use 'orca job run <spec.md>' to submit one.")
return nil
}
fmt.Fprintf(cmd.OutOrStdout(), "%-36s %-20s %-12s %-8s\n", "ID", "NAME", "STATUS", "EXIT")
fmt.Fprintf(cmd.OutOrStdout(), "%-10s %-20s %-12s %-20s %-5s\n", "ID", "NAME", "STATUS", "NODE", "EXIT")
for _, j := range jobs {
fmt.Fprintf(cmd.OutOrStdout(), "%-36s %-20s %-12s %-8d\n", j.ID, j.Name, j.Status, j.ExitCode)
shortID := j.ID
if len(shortID) > 8 {
shortID = shortID[:8]
}
exit := "-"
if j.Status == model.JobStatusComplete || j.Status == model.JobStatusFailed || j.Status == model.JobStatusStopped {
exit = fmt.Sprintf("%d", j.ExitCode)
}
fmt.Fprintf(cmd.OutOrStdout(), "%-10s %-20s %-12s %-20s %-5s\n", shortID, j.Name, j.Status, j.Node, exit)
}
return nil
}
@@ -286,9 +298,17 @@ func renderJobTable(jobs []*model.Job) string {
if len(jobs) == 0 {
return "No jobs.\n"
}
out := fmt.Sprintf("%-36s %-20s %-12s %-8s\n", "ID", "NAME", "STATUS", "EXIT")
out := fmt.Sprintf("%-10s %-20s %-12s %-20s %-5s\n", "ID", "NAME", "STATUS", "NODE", "EXIT")
for _, j := range jobs {
out += fmt.Sprintf("%-36s %-20s %-12s %-8d\n", j.ID, j.Name, j.Status, j.ExitCode)
shortID := j.ID
if len(shortID) > 8 {
shortID = shortID[:8]
}
exit := "-"
if j.Status == model.JobStatusComplete || j.Status == model.JobStatusFailed || j.Status == model.JobStatusStopped {
exit = fmt.Sprintf("%d", j.ExitCode)
}
out += fmt.Sprintf("%-10s %-20s %-12s %-20s %-5s\n", shortID, j.Name, j.Status, j.Node, exit)
}
return out
}
@@ -577,3 +597,23 @@ func splitCommand(s string) (string, []string) {
}
return parts[0], parts[1:]
}
// insertRemoteJob inserts a model.Job row for a remotely-deployed job
// (REQ-166, Phase C2). Without this, `job list` shows nothing for remote
// deployments and `job stop` can't find the node.
func insertRemoteJob(spec *jobspec.WorkloadSpec, node string) error {
db, closer, err := openDB()
if err != nil {
return err
}
defer closer()
repo := store.NewJobRepo(db)
return repo.Insert(context.Background(), &model.Job{
ID: uuid.NewString(),
Name: spec.Name,
Spec: "",
Status: model.JobStatusRunning,
CreatedAt: time.Now().UTC(),
Node: node,
})
}
+101 -19
View File
@@ -34,6 +34,7 @@ import (
"git.cloudinit.dev/coreci/orca/internal/emitter"
"git.cloudinit.dev/coreci/orca/internal/jobspec"
"git.cloudinit.dev/coreci/orca/internal/model"
"git.cloudinit.dev/coreci/orca/internal/runtime"
"git.cloudinit.dev/coreci/orca/internal/scheduler"
"git.cloudinit.dev/coreci/orca/internal/sshpush"
"git.cloudinit.dev/coreci/orca/internal/store"
@@ -199,30 +200,72 @@ func deployRemote(ctx context.Context, spec *jobspec.WorkloadSpec, res *dispatch
return nil, fmt.Errorf("deployRemote: selected node %q not found in registry", res.node)
}
// Render the systemd unit via the emitter. The runtime is required
// for the process emitter; a spec with no runtime has nothing to
// ExecStart and is rejected by the emitter.
// REQ-166 / Phase C3: branch on runtime + node kind.
runtimeOneOf := ""
if spec.Runtime != nil {
runtimeOneOf = spec.Runtime.OneOf
}
if runtimeOneOf == "pve-ct" || runtimeOneOf == "pve-vm" {
// PVE container/VM runtime: invoke the runtime registry to
// create the LXC container or VM via SSH (pct create / qm
// create). Only valid on proxmox nodes.
if node.Kind != string(model.NodeKindProxmox) {
return nil, fmt.Errorf("deployRemote: runtime %q requires a proxmox node (node %q is %q)", runtimeOneOf, res.node, node.Kind)
}
peer := sshPeerFor(node)
sshTransport, err := newSSHPushTransport()
if err != nil {
return nil, fmt.Errorf("deployRemote: transport: %w", err)
}
defer sshTransport.Close()
alloc := &runtime.Alloc{
ID: res.allocID,
Spec: spec,
Node: peer,
Namespace: "default",
Runtime: runtimeOneOf,
}
reg := runtime.DefaultRegistry(sshTransport)
if err := reg.Prepare(ctx, alloc); err != nil {
return nil, fmt.Errorf("deployRemote: pve prepare: %w", err)
}
if _, err := reg.Start(ctx, alloc); err != nil {
return nil, fmt.Errorf("deployRemote: pve start: %w", err)
}
// For PVE workloads, also emit Traefik route if the spec has ports.
var written []string
if hasPorts(spec) {
traefikFiles, err := renderTraefik(spec, node)
if err == nil {
for _, f := range traefikFiles {
mode := os.FileMode(0o644)
_ = sshTransport.WriteFile(ctx, peer, f.Path, []byte(f.Content), mode)
written = append(written, f.Path)
}
}
}
return written, nil
}
// Process runtime: systemd units (only on linux/localhost nodes).
if node.Kind == string(model.NodeKindProxmox) {
return nil, fmt.Errorf("deployRemote: runtime %q requires a linux node (node %q is proxmox; use one_of: pve-ct or pve-vm for proxmox)", runtimeOneOf, res.node)
}
// Render the systemd unit via the emitter.
em := emitter.SystemdEmitter{}
enode := &emitter.Node{
Hostname: node.Name,
Runtime: []string{"process"},
Tags: nil,
}
// Advertise the node kind as a runtime so the emitter can branch
// (proxmox nodes expose pve-* runtimes). For process workloads
// this is informational.
if node.Kind == string(model.NodeKindProxmox) {
enode.Runtime = append(enode.Runtime, "proxmox")
}
files, err := em.Render(spec, enode)
if err != nil {
return nil, fmt.Errorf("deployRemote: render unit: %w", err)
}
// T9: systemd-analyze verify on the rendered unit before deploy.
// Run it locally (the unit is a portable text file); if
// systemd-analyze is not installed, skip silently (dev boxes
// without systemd). A verification FAILURE is an error.
for _, f := range files {
if err := verifySystemdUnit(ctx, f.Path, f.Content); err != nil {
return nil, fmt.Errorf("deployRemote: systemd-analyze verify %s: %w", f.Path, err)
@@ -241,24 +284,18 @@ func deployRemote(ctx context.Context, spec *jobspec.WorkloadSpec, res *dispatch
for _, f := range files {
mode := os.FileMode(0o644)
if f.Mode != "" {
// f.Mode is an octal string like "0644".
var m uint64
if _, perr := fmt.Sscanf(f.Mode, "%o", &m); perr == nil {
mode = os.FileMode(m)
}
}
if err := transport.WriteFile(ctx, peer, f.Path, []byte(f.Content), mode); err != nil {
// C-44: SSH-push failure -> error, NOT local fallback.
return nil, fmt.Errorf("deployRemote: push %s to %s (%s): %w", f.Path, res.node, peer, err)
}
written = append(written, f.Path)
}
// Reload systemd + enable the unit so it starts at boot. These are
// best-effort; a failure here is surfaced but does not undo the
// push (the unit is on disk). We use systemctl daemon-reload +
// enable --now for each .service unit (.target units for task
// groups are also enabled).
// Reload systemd + enable the unit so it starts at boot.
for _, p := range written {
if !strings.HasSuffix(p, ".service") && !strings.HasSuffix(p, ".target") {
continue
@@ -268,9 +305,46 @@ func deployRemote(ctx context.Context, spec *jobspec.WorkloadSpec, res *dispatch
}
}
// REQ-166 / Phase C4: emit Traefik dynamic config if the spec
// has ports (is a Service with ingress).
if hasPorts(spec) {
traefikFiles, err := renderTraefik(spec, node)
if err != nil {
// Non-fatal: Traefik route is best-effort.
return written, nil
}
for _, f := range traefikFiles {
mode := os.FileMode(0o644)
_ = transport.WriteFile(ctx, peer, f.Path, []byte(f.Content), mode)
written = append(written, f.Path)
}
}
return written, nil
}
// hasPorts returns true if the spec declares any ports (is a Service).
func hasPorts(spec *jobspec.WorkloadSpec) bool {
if spec == nil {
return false
}
if len(spec.Ports) > 0 {
return true
}
return false
}
// renderTraefik renders the Traefik dynamic config for the spec + node.
func renderTraefik(spec *jobspec.WorkloadSpec, node *model.Node) ([]emitter.File, error) {
em := emitter.TraefikEmitter{}
enode := &emitter.Node{
Hostname: node.Name,
Runtime: []string{"process"},
Tags: nil,
}
return em.Render(spec, enode)
}
// verifySystemdUnit runs `systemd-analyze verify` on the rendered unit
// content. The unit is written to a temp file (with its real basename)
// so systemd-analyze resolves fragment paths correctly. When
@@ -433,3 +507,11 @@ func logDispatch(res *dispatchResult, err error) {
}
log.Info("job.dispatch", attrs...)
}
// newSSHPushTransport creates a concrete sshpush.Transport for PVE
// runtime operations (pct create/qm create). The jobDispatchTransport
// interface wraps sshpush.Transport but the runtime package needs the
// concrete type.
func newSSHPushTransport() (*sshpush.Transport, error) {
return sshpush.NewTransport(certpaths.SSHKeyPath(), certpaths.KnownHostsPath()), nil
}
+6 -3
View File
@@ -55,6 +55,7 @@ var (
joinSSHKey string
joinSSHPort int
joinHostKeyFP string
joinLXCTemplate string
proxmoxUser string
proxmoxRole string
leaveID string
@@ -186,6 +187,7 @@ func joinProxmox(cmd *cobra.Command) error {
SSHPort: joinSSHPort,
HostKeyFingerprint: joinHostKeyFP,
Logger: newLogger(),
LXCTemplate: joinLXCTemplate,
})
if err != nil {
return fmt.Errorf("proxmox bootstrap: %w", err)
@@ -250,7 +252,7 @@ func joinLinux(cmd *cobra.Command) error {
SSHPort: joinSSHPort,
HostKeyFingerprint: joinHostKeyFP,
Logger: newLogger(),
})
})
if err != nil {
return fmt.Errorf("linux bootstrap: %w", err)
}
@@ -503,14 +505,15 @@ func init() {
nodeJoinCmd.Flags().StringVar(&joinName, "name", "", "node name (required for --type localhost)")
nodeJoinCmd.Flags().StringVar(&joinAddr, "addr", "", "node address (default localhost:8443)")
nodeJoinCmd.Flags().StringVar(&joinCAFinger, "ca-fingerprint", "", "pin CA cert SHA-256 (REQ-026); fails if on-disk CA doesn't match")
nodeJoinCmd.Flags().StringVar(&joinType, "type", "localhost", "node type: localhost (default) or proxmox (SSH bootstrap)")
nodeJoinCmd.Flags().StringVar(&joinType, "type", "localhost", "node type: localhost (default), proxmox, or linux (SSH bootstrap)")
nodeJoinCmd.Flags().StringVar(&joinHost, "host", "", "proxmox host address (IP/hostname, no port; required for --type proxmox)")
nodeJoinCmd.Flags().StringVar(&joinSSHUser, "ssh-user", "root", "SSH username for proxmox bootstrap (default root)")
nodeJoinCmd.Flags().StringVar(&joinSSHKey, "ssh-key", "", "SSH private key path for proxmox bootstrap (R-021: no passwords; default: orca key)")
nodeJoinCmd.Flags().IntVar(&joinSSHPort, "ssh-port", 22, "SSH port for proxmox bootstrap (default 22)")
nodeJoinCmd.Flags().StringVar(&proxmoxUser, "proxmox-user", "orca", "Linux system user to create on the proxmox host (config-overridable)")
nodeJoinCmd.Flags().StringVar(&proxmoxRole, "proxmox-role", "OrcaOperator", "PVE custom role to create (config-overridable)")
nodeJoinCmd.Flags().StringVar(&joinHostKeyFP, "host-key-fingerprint", "", "SSH host key SHA256:base64 fingerprint (pre-pin; supersedes TOFU for --type proxmox)")
nodeJoinCmd.Flags().StringVar(&joinHostKeyFP, "host-key-fingerprint", "", "SSH host key SHA256:base64 fingerprint (pre-pin; supersedes TOFU for --type proxmox or --type linux)")
nodeJoinCmd.Flags().StringVar(&joinLXCTemplate, "lxc-template", "ubuntu-24.04", "LXC template for Proxmox (default ubuntu-24.04; alternatives: alpine-3.20, debian-12)")
nodeLeaveCmd.Flags().StringVar(&leaveID, "id", "", "node id")
nodeListCmd.Flags().BoolVar(&nodeWatch, "watch", false, "stream nodes until Ctrl-C (table refresh or --json per-event)")
+116 -6
View File
@@ -11,10 +11,14 @@ package cli
import (
"context"
"fmt"
"strconv"
"strings"
"time"
"github.com/spf13/cobra"
"git.cloudinit.dev/coreci/orca/internal/certpaths"
"git.cloudinit.dev/coreci/orca/internal/sshpush"
"git.cloudinit.dev/coreci/orca/internal/store"
)
@@ -72,8 +76,11 @@ var nodeCapacitySetCmd = &cobra.Command{
Short: "Declare capacity for a node (used by bin-packing)",
Long: "Write cpu_millicores, memory_mib, and disk_mib for the named node. Idempotent: subsequent calls overwrite.",
RunE: func(cmd *cobra.Command, args []string) error {
if capSetCPU <= 0 || capSetMem <= 0 || capSetDisk <= 0 {
return fmt.Errorf("--cpu, --memory, and --disk must all be positive")
// REQ-168: allow partial updates. At least one dimension
// must be positive; the others are read from the existing
// row (or default to 0 if no row exists yet).
if capSetCPU <= 0 && capSetMem <= 0 && capSetDisk <= 0 {
return fmt.Errorf("at least one of --cpu, --memory, or --disk must be positive")
}
id := capNodeID
if id == "" {
@@ -87,11 +94,27 @@ var nodeCapacitySetCmd = &cobra.Command{
}
defer closer()
repo := store.NewCapacityRepo(db)
// Read existing row for partial update.
existing, _ := repo.Get(ctx, id)
cpu := capSetCPU
mem := capSetMem
disk := capSetDisk
if existing != nil {
if cpu <= 0 {
cpu = existing.CPUMillicores
}
if mem <= 0 {
mem = existing.MemoryMiB
}
if disk <= 0 {
disk = existing.DiskMiB
}
}
c := &store.NodeCapacity{
NodeID: id,
CPUMillicores: capSetCPU,
MemoryMiB: capSetMem,
DiskMiB: capSetDisk,
CPUMillicores: cpu,
MemoryMiB: mem,
DiskMiB: disk,
}
if err := repo.Upsert(ctx, c); err != nil {
return err
@@ -137,6 +160,92 @@ var nodeCapacityListCmd = &cobra.Command{
},
}
// nodeCapacityAutoCmd discovers capacity by SSHing to the node and
// reading nproc, /proc/meminfo, df (REQ-168, Phase D2).
var capAutoPct int
var nodeCapacityAutoCmd = &cobra.Command{
Use: "auto [percentage]",
Short: "Auto-discover node capacity via SSH (default 75% of physical)",
Long: `SSH to the specified node and discover CPU cores, memory,
and disk capacity. Multiplies the physical values by the given
percentage (default 75) to reserve headroom for the OS. The discovered
values are written to the capacity table (same as 'orca node capacity set').`,
Args: cobra.MaximumNArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
pct := 75
if len(args) > 0 {
var err error
pct, err = strconv.Atoi(args[0])
if err != nil || pct < 1 || pct > 100 {
return fmt.Errorf("percentage must be 1-100, got %q", args[0])
}
}
id := capNodeID
if id == "" {
return fmt.Errorf("--node is required for capacity auto")
}
// Build SSH transport and exec discovery commands.
transport := sshpush.NewTransport(certpaths.SSHKeyPath(), certpaths.KnownHostsPath())
defer transport.Close()
ctx, cancel := context.WithTimeout(cmd.Context(), 30*time.Second)
defer cancel()
// CPU: nproc
cpuOut, err := transport.Exec(ctx, id, "nproc")
if err != nil {
return fmt.Errorf("capacity auto: SSH exec nproc on %s: %w", id, err)
}
cores, err := strconv.Atoi(strings.TrimSpace(string(cpuOut)))
if err != nil {
return fmt.Errorf("capacity auto: parse nproc output %q: %w", string(cpuOut), err)
}
// Memory: MemTotal from /proc/meminfo (in kB -> MiB)
memOut, err := transport.Exec(ctx, id, "awk '/MemTotal/{print $2}' /proc/meminfo")
if err != nil {
return fmt.Errorf("capacity auto: SSH exec meminfo on %s: %w", id, err)
}
memKB, err := strconv.ParseInt(strings.TrimSpace(string(memOut)), 10, 64)
if err != nil {
return fmt.Errorf("capacity auto: parse meminfo output %q: %w", string(memOut), err)
}
// Disk: df on root (1K-blocks -> MiB)
diskOut, err := transport.Exec(ctx, id, "df --output=size / | tail -1")
if err != nil {
return fmt.Errorf("capacity auto: SSH exec df on %s: %w", id, err)
}
diskKB, err := strconv.ParseInt(strings.TrimSpace(string(diskOut)), 10, 64)
if err != nil {
return fmt.Errorf("capacity auto: parse df output %q: %w", string(diskOut), err)
}
// Apply percentage, convert to millicores/MiB.
cpuM := int64(cores) * 1000 * int64(pct) / 100
memMib := memKB * int64(pct) / 100 / 1024
diskMib := diskKB * int64(pct) / 100 / 1024
// Write to DB.
db, closer, err := openDB()
if err != nil {
return err
}
defer closer()
repo := store.NewCapacityRepo(db)
c := &store.NodeCapacity{
NodeID: id,
CPUMillicores: cpuM,
MemoryMiB: memMib,
DiskMiB: diskMib,
}
if err := repo.Upsert(ctx, c); err != nil {
return err
}
if jsonOutput {
return printJSON(c)
}
fmt.Fprintf(cmd.OutOrStdout(), "Capacity auto-discovered for %s (%d%%): cpu=%dm, mem=%dMiB, disk=%dMiB\n", id, pct, cpuM, memMib, diskMib)
return nil
},
}
func init() {
nodeCapacitySetCmd.Flags().Int64Var(&capSetCPU, "cpu", 0, "CPU capacity in millicores (1000 = 1 vCPU)")
nodeCapacitySetCmd.Flags().Int64Var(&capSetMem, "memory", 0, "Memory capacity in MiB")
@@ -144,6 +253,7 @@ func init() {
nodeCapacitySetCmd.Flags().StringVar(&capNodeID, "node", "", "node id (defaults to 'self')")
nodeCapacityShowCmd.Flags().StringVar(&capNodeID, "node", "", "node id (defaults to 'self')")
nodeCapacityCmd.AddCommand(nodeCapacityShowCmd, nodeCapacitySetCmd, nodeCapacityListCmd)
nodeCapacityAutoCmd.Flags().StringVar(&capNodeID, "node", "", "node name or ID to auto-discover capacity for")
nodeCapacityCmd.AddCommand(nodeCapacityShowCmd, nodeCapacitySetCmd, nodeCapacityListCmd, nodeCapacityAutoCmd)
nodeCmd.AddCommand(nodeCapacityCmd)
}
+5 -3
View File
@@ -10,16 +10,18 @@ import (
"git.cloudinit.dev/coreci/orca/internal/store"
)
func TestNodeCapacitySetMissingArgs(t *testing.T) {
func TestNodeCapacitySetPartialUpdate(t *testing.T) {
_, cleanup := initTestEnv(t)
defer cleanup()
resetRootFlags(t)
var buf bytes.Buffer
rootCmd.SetOut(&buf)
rootCmd.SetErr(&buf)
// REQ-168: partial updates are now allowed. Setting only --cpu
// should succeed (memory/disk default to 0 or existing values).
rootCmd.SetArgs([]string{"node", "capacity", "set", "--cpu", "1000"})
if err := rootCmd.Execute(); err == nil {
t.Fatal("expected error for capacity set missing memory/disk, got nil")
if err := rootCmd.Execute(); err != nil {
t.Fatalf("expected success for partial capacity set, got: %v", err)
}
}
+11
View File
@@ -0,0 +1,11 @@
package cli
import (
"git.cloudinit.dev/coreci/orca/internal/traefik"
)
var traefikVersion = traefik.DefaultVersion
func installTraefikLocal() error {
return traefik.InstallLocal(traefikVersion)
}
+2 -2
View File
@@ -111,8 +111,8 @@ func TestWatchJobs_TableRefresh(t *testing.T) {
if !strings.Contains(output, "\033[2J\033[H") {
t.Errorf("expected clear-screen escape in table watch output, got: %s", output)
}
if !strings.Contains(output, "table-job") {
t.Errorf("expected table-job in output, got: %s", output)
if !strings.Contains(output, "table-jo") {
t.Errorf("expected table-jo in output, got: %s", output)
}
}
+24 -15
View File
@@ -32,6 +32,7 @@ import (
"git.cloudinit.dev/coreci/orca/internal/osdetect"
"git.cloudinit.dev/coreci/orca/internal/proxmox"
"git.cloudinit.dev/coreci/orca/internal/security"
"git.cloudinit.dev/coreci/orca/internal/sshpush"
"git.cloudinit.dev/coreci/orca/internal/store"
"git.cloudinit.dev/coreci/orca/internal/transport"
)
@@ -225,22 +226,22 @@ func DB() Check {
}
}
// Network probes peer reachability via mTLS /healthz (REQ-032 completion).
// Peers are sourced from the persisted nodes table (not the in-memory
// PeerRegistry, which is empty at CLI time). Zero peers → WARN (single-node
// is legitimate). Any peer unreachable → FAIL (D-038).
// Network probes peer reachability via SSH exec (REQ-164 Phase A5).
// The SSH-push model (R-001) has no daemon on :8443, so the HTTP /healthz
// probe is replaced with an SSH "echo ok" exec. Peers are sourced from
// the persisted nodes table. Zero peers → WARN (single-node is
// legitimate). Any peer unreachable → FAIL (D-038).
func Network() Check {
return Check{
Name: "network",
Description: "peer reachability via mTLS /healthz probe",
Description: "peer reachability via SSH exec probe",
Run: func(ctx context.Context) (Result, string) {
caPath := certpaths.CACertPath()
certPath := certpaths.ServerCertPath()
keyPath := certpaths.ServerKeyPath()
keyPath := certpaths.SSHKeyPath()
khPath := certpaths.KnownHostsPath()
// Check that cert files exist before attempting probes.
if _, err := os.Stat(caPath); err != nil {
return ResultFail, fmt.Sprintf("CA cert missing: %v (run `orca cert init`)", err)
// Check that the SSH key exists.
if _, err := os.Stat(keyPath); err != nil {
return ResultFail, fmt.Sprintf("SSH key missing: %v (run `orca init`)", err)
}
path := certpaths.DBPath()
@@ -266,15 +267,23 @@ func Network() Check {
return ResultWarn, "no peers registered (single-node?)"
}
// For localhost nodes, check SSH to 127.0.0.1:22 (may fail if
// SSH isn't running — that's OK, report WARN not FAIL).
transport := sshpush.NewTransport(keyPath, khPath)
var lines []string
anyFail := false
for _, n := range live {
probeCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
err := probeHealthz(probeCtx, caPath, certPath, keyPath, n.Name, n.Address)
probeCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
_, err := transport.Exec(probeCtx, n.Name, "echo ok")
cancel()
if err != nil {
anyFail = true
lines = append(lines, fmt.Sprintf(" %s (%s): %v", n.Name, n.Address, err))
if n.Kind == string(model.NodeKindLocalhost) {
lines = append(lines, fmt.Sprintf(" %s (%s): %v (SSH to self may not be running)", n.Name, n.Address, err))
} else {
anyFail = true
lines = append(lines, fmt.Sprintf(" ✗ %s (%s): %v", n.Name, n.Address, err))
}
} else {
lines = append(lines, fmt.Sprintf(" ✓ %s (%s)", n.Name, n.Address))
}
+12 -2
View File
@@ -99,6 +99,10 @@ func TestRunWithCAAndServerCert(t *testing.T) {
if err := security.WriteKey(dir+"/server.key", keyPEM); err != nil {
t.Fatalf("WriteKey: %v", err)
}
// REQ-164: network check requires the SSH key to exist.
if _, _, err := security.GenerateOrLoadSSHKey(dir); err != nil {
t.Fatalf("GenerateOrLoadSSHKey: %v", err)
}
rep := Run(context.Background())
byName := make(map[string]CheckResult, len(rep.Checks))
@@ -152,6 +156,9 @@ func TestDBCheck_IntegrityOK(t *testing.T) {
func TestNetworkCheck_NoPeers(t *testing.T) {
dir := t.TempDir()
t.Setenv("ORCA_HOME", dir)
// REQ-164: network check requires SSH key.
_, _, _ = security.GenerateOrLoadSSHKey(dir)
t.Setenv("ORCA_DB", filepath.Join(dir, "orca.db"))
// Create a CA + server cert so the network check can build a client.
@@ -179,6 +186,9 @@ func TestNetworkCheck_NoPeers(t *testing.T) {
func TestNetworkCheck_PeerUnreachable(t *testing.T) {
dir := t.TempDir()
t.Setenv("ORCA_HOME", dir)
// REQ-164: network check requires SSH key.
_, _, _ = security.GenerateOrLoadSSHKey(dir)
t.Setenv("ORCA_DB", filepath.Join(dir, "orca.db"))
// Create a CA + server cert.
@@ -225,8 +235,8 @@ func TestNetworkCheck_NoCert(t *testing.T) {
if r != ResultFail {
t.Errorf("Network check: got %s, want FAIL — %s", r, msg)
}
if !strings.Contains(msg, "CA cert missing") {
t.Errorf("Network check message should mention missing CA, got: %s", msg)
if !strings.Contains(msg, "SSH key missing") && !strings.Contains(msg, "CA cert missing") {
t.Errorf("Network check message should mention missing key/cert, got: %s", msg)
}
}
+3 -1
View File
@@ -124,6 +124,8 @@ func RegisterTraefik(reg *Registry) {
reg.Register("service:process", e)
reg.Register("service:podman", e)
reg.Register("service:wasm", e)
reg.Register("service:pve-ct", e)
reg.Register("service:pve-vm", e)
}
// renderTraefikYAML renders the Traefik dynamic-config YAML for the
@@ -288,7 +290,7 @@ func renderTraefikStaticYAML(o TraefikStaticOpts) string {
b.WriteString(fmt.Sprintf(" address: %q\n", "127.0.0.1:8081"))
b.WriteString("\nproviders:\n")
b.WriteString(" file:\n")
b.WriteString(fmt.Sprintf(" filename: %q\n", "/etc/traefik/dynamic/orca.yml"))
b.WriteString(fmt.Sprintf(" directory: %q\n", traefikDynamicDir))
b.WriteString(" watch: true\n")
b.WriteString("\nlog:\n")
b.WriteString(" level: INFO\n")
+1 -1
View File
@@ -375,7 +375,7 @@ func TestTraefikEmitter_RenderStaticConfigHybrid(t *testing.T) {
`address: "127.0.0.1:8081"`,
"providers:",
"file:",
`filename: "/etc/traefik/dynamic/orca.yml"`,
`directory: "/etc/traefik/dynamic"`,
"watch: true",
"log:",
"level: INFO",
+1 -1
View File
@@ -152,7 +152,7 @@ func (d *Dispatcher) dispatchTo(ctx context.Context, targetNode string, specByte
return d.dispatchToPeer(ctx, p, specBytes, idempotencyKey)
}
}
return "", "", fmt.Errorf("dispatchTo: target node %q not found in peer registry", targetNode)
return "", "", fmt.Errorf("dispatchTo: target node %q not found in peer registry (looked up by ID and name)", targetNode)
}
// dispatchToPeer opens an mTLS client and calls Submit on the peer.
+23 -9
View File
@@ -30,10 +30,11 @@ import (
"time"
"golang.org/x/crypto/ssh"
"golang.org/x/crypto/ssh/knownhosts"
"git.cloudinit.dev/coreci/orca/internal/certpaths"
"git.cloudinit.dev/coreci/orca/internal/proxmox"
"git.cloudinit.dev/coreci/orca/internal/security"
"git.cloudinit.dev/coreci/orca/internal/traefik"
)
// DefaultSSHUser is the default SSH username for the initial connection.
@@ -112,17 +113,16 @@ func BootstrapLinux(ctx context.Context, opts Options) (*Result, error) {
}
hostKeyCallback = hkcb
} else {
hkcb, err := knownhosts.New(certpaths.KnownHostsPath())
// REQ-164 / Phase A3: reuse the tested Proxmox TOFU callback
// which handles first-connect key capture + known_hosts file
// creation (create-on-open). The previous inline implementation
// failed on first connect with a raw KeyError because it never
// wrote the captured key.
hkcb, err := proxmox.TOFUHostKeyCallbackPath(certpaths.KnownHostsPath(), sshAddr, &capturedHostKey)
if err != nil {
return nil, fmt.Errorf("linux bootstrap: known_hosts: %w", err)
}
hostKeyCallback = ssh.HostKeyCallback(func(hostname string, remote net.Addr, key ssh.PublicKey) error {
err := hkcb(hostname, remote, key)
if err == nil {
capturedHostKey = key
}
return err
})
hostKeyCallback = hkcb
}
sshConfig := &ssh.ClientConfig{
@@ -157,6 +157,20 @@ func BootstrapLinux(ctx context.Context, opts Options) (*Result, error) {
}
opts.Logger.Info("linux bootstrap: user created", "user", opts.OrcaUser)
// Step 4d: Install Traefik on the remote host (REQ-165, Phase B).
// Traefik is the data-plane ingress; SSH is control plane only.
sshExecFn := func(cmd string) ([]byte, error) {
session, err := client.NewSession()
if err != nil {
return nil, err
}
defer session.Close()
return session.CombinedOutput(cmd)
}
if err := traefik.InstallRemote("", sshExecFn); err != nil {
opts.Logger.Warn("linux bootstrap: traefik install failed", "err", err)
}
// Step 5: Create the drift-events directory.
if err := sshExec(client, fmt.Sprintf(
"mkdir -p ~%s/drift-events && chown %s:%s ~%s/drift-events",
+2
View File
@@ -21,6 +21,7 @@ type Job struct {
StartedAt *time.Time `json:"started_at,omitempty"`
EndedAt *time.Time `json:"ended_at,omitempty"`
ExitCode int `json:"exit_code"`
Node string `json:"node,omitempty"`
}
type TaskStatus string
@@ -41,6 +42,7 @@ type Task struct {
Env []string `json:"env,omitempty"`
PID int `json:"pid"`
ExitCode int `json:"exit_code"`
Node string `json:"node,omitempty"`
Status TaskStatus `json:"status"`
CreatedAt time.Time `json:"created_at"`
StartedAt *time.Time `json:"started_at,omitempty"`
+31
View File
@@ -38,6 +38,7 @@ import (
"git.cloudinit.dev/coreci/orca/internal/certpaths"
"git.cloudinit.dev/coreci/orca/internal/security"
"git.cloudinit.dev/coreci/orca/internal/traefik"
)
// DefaultProxmoxUser is the default Linux system user created on the
@@ -82,6 +83,9 @@ type Options struct {
HostKeyFingerprint string
// Logger receives audit-log entries. If nil, slog.Default() is used.
Logger *slog.Logger
// LXCTemplate is the LXC template to download during bootstrap
// (default "ubuntu-24.04"; alternatives: "alpine-3.20", "debian-12").
LXCTemplate string
}
// Result is the outcome of a successful bootstrap.
@@ -243,6 +247,21 @@ func BootstrapProxmox(ctx context.Context, opts Options) (*Result, error) {
return nil, fmt.Errorf("validate sudoers: %w", err)
}
// Step 9a: Install Traefik on the Proxmox host (REQ-165, Phase B).
// Traefik runs on the PVE OS as the data-plane ingress; SSH is
// control plane only. Idempotent: skips if binary already exists.
if err := traefik.InstallRemote("", runRemote); err != nil {
log.Warn("proxmox.traefik_install_failed", "err", err)
}
// Step 9b: Download default LXC template (REQ-167, Phase C).
// Default: ubuntu-24.04. Configurable via --lxc-template.
template := opts.LXCTemplate
if template == "" {
template = "ubuntu-24.04"
}
_, _ = runRemote(fmt.Sprintf("pveam download local %s 2>/dev/null || true", shellQuote(template)))
log.Info("proxmox.bootstrap_ok",
slog.String("event", "proxmox.bootstrap_ok"),
slog.String("host", opts.Host),
@@ -318,6 +337,18 @@ func TOFUHostKeyCallbackPath(knownHostsPath, addr string, capturedKey *ssh.Publi
if knownHostsPath == "" {
knownHostsPath = certpaths.KnownHostsPath()
}
// REQ-164 / Phase A2: create the known_hosts file if it doesn't
// exist (knownhosts.New requires the file to be present). This is
// defense-in-depth alongside init.go which also creates it.
if _, err := os.Stat(knownHostsPath); err != nil {
if os.IsNotExist(err) {
if writeErr := security.WriteAtomic(knownHostsPath, 0o600, []byte{}); writeErr != nil {
return nil, fmt.Errorf("tofu create known_hosts: %w", writeErr)
}
} else {
return nil, fmt.Errorf("tofu stat known_hosts: %w", err)
}
}
cb, err := knownhosts.New(knownHostsPath)
if err != nil {
return nil, err
+78
View File
@@ -0,0 +1,78 @@
package traefik
import (
"fmt"
"os/exec"
"strings"
)
const DefaultVersion = "v3.3.0"
func downloadURL(version string) string {
return fmt.Sprintf("https://github.com/traefik/traefik/releases/download/%s/traefik_%s_linux_amd64.tar.gz", version, version)
}
func InstallLocal(version string) error {
if version == "" {
version = DefaultVersion
}
if _, err := exec.LookPath("traefik"); err == nil {
ensureDirs()
return nil
}
url := downloadURL(version)
cmd := exec.Command("bash", "-c",
fmt.Sprintf(`curl -fsSL %s | tar -xzf - -C /usr/local/bin/ traefik && chmod +x /usr/local/bin/traefik`, url))
if out, err := cmd.CombinedOutput(); err != nil {
return fmt.Errorf("download traefik %s: %w (output: %s)", version, err, string(out))
}
ensureDirs()
writeSystemdUnit()
_ = exec.Command("systemctl", "daemon-reload").Run()
_ = exec.Command("systemctl", "enable", "--now", "orca-traefik").Run()
return nil
}
type RemoteExecFunc func(cmd string) ([]byte, error)
func InstallRemote(version string, execFn RemoteExecFunc) error {
if version == "" {
version = DefaultVersion
}
if out, err := execFn("command -v traefik"); err == nil && len(strings.TrimSpace(string(out))) > 0 {
_, _ = execFn("mkdir -p /etc/traefik/dynamic")
return nil
}
url := downloadURL(version)
installCmd := fmt.Sprintf(`curl -fsSL %s | tar -xzf - -C /usr/local/bin/ traefik && chmod +x /usr/local/bin/traefik && mkdir -p /etc/traefik/dynamic`, url)
if out, err := execFn(installCmd); err != nil {
return fmt.Errorf("download traefik on remote: %w (output: %s)", err, string(out))
}
unit := systemdUnitContent()
_, _ = execFn(fmt.Sprintf(`echo '%s' > /etc/systemd/system/orca-traefik.service`, unit))
_, _ = execFn("systemctl daemon-reload && systemctl enable --now orca-traefik")
return nil
}
func ensureDirs() {
_ = exec.Command("mkdir", "-p", "/etc/traefik/dynamic").Run()
}
func writeSystemdUnit() {
_ = exec.Command("bash", "-c", fmt.Sprintf(`echo '%s' > /etc/systemd/system/orca-traefik.service`, systemdUnitContent())).Run()
}
func systemdUnitContent() string {
return `[Unit]
Description=Orca Traefik Data Plane
After=network.target
[Service]
Type=simple
ExecStart=/usr/local/bin/traefik --configFile=/etc/traefik/traefik.yml
Restart=on-failure
RestartSec=5s
[Install]
WantedBy=multi-user.target`
}