internal/agent/run/cli.go
Ref: Size: 17.0 KiB History
// cli.go is the eitri-agent command line: `eitri-agent join <blob>` enrolls the
// host once, and `eitri-agent` (no subcommand) runs the reconcile + sync loop
// indefinitely. It lives here rather than in cmd/eitri-agent so the flag/config
// assembly, validation, and dispatch are testable and coverage-gated (arch R14:
// main packages are wiring only).
// Package run implements the eitri-agent command line behind a tested RunCLI so
// cmd/eitri-agent stays thin wiring (arch R14).
package run
import (
"context"
"errors"
"flag"
"fmt"
"log/slog"
"net"
"os"
"os/exec"
"os/signal"
"runtime"
"strings"
"syscall"
"time"
"github.com/a73x/eitri/internal/agent/enrollclient"
"github.com/a73x/eitri/internal/agent/exposeproxy"
"github.com/a73x/eitri/internal/agent/hostinfo"
"github.com/a73x/eitri/internal/agent/imagecache"
"github.com/a73x/eitri/internal/agent/reconcile"
"github.com/a73x/eitri/internal/agent/seed"
"github.com/a73x/eitri/internal/agent/state"
"github.com/a73x/eitri/internal/agent/statelock"
"github.com/a73x/eitri/internal/agent/syncclient"
"github.com/a73x/eitri/internal/covsnap"
"github.com/a73x/eitri/internal/joinblob"
"github.com/a73x/eitri/internal/names"
"github.com/a73x/eitri/internal/version"
)
// hostRunner is the production one-shot command runner injected into the
// host-touching agent packages — which ones exist depends on the platform. It
// lives in the composition root because it is a wiring value: constructing the
// concrete dependency is what a root is for, and keeping it here means no
// platform's wiring has to reach into another platform's provisioner for it.
func hostRunner(ctx context.Context, name string, args ...string) (string, error) {
out, err := exec.CommandContext(ctx, name, args...).CombinedOutput()
return string(out), err
}
// Config carries serve's wiring, replacing a long positional list.
type Config struct {
StateDir, CHBin, Firmware string
VfkitBin string
BridgeCIDR string
BootstrapURL string
TombstoneGrace, VanishGrace time.Duration
VMTimeout time.Duration
ImageCacheMaxGB int64
// MaxConcurrentCreates bounds how much of the per-VM workers' create
// concurrency reaches the disk at once (0 = unlimited).
MaxConcurrentCreates int
// MaxVCPUs, MaxMemMB, MaxDiskGB cap the host resources this agent offers the
// fleet (0 = unlimited): advertised to the server AND enforced at VM boot.
MaxVCPUs, MaxMemMB, MaxDiskGB int64
// HostNetworks maps a named guest network to the bridge this host's
// operator already declared for it (Linux hosts only). eitri attaches taps
// to the bridge; it never creates or addresses it.
HostNetworks map[string]string
}
// RunCLI dispatches the eitri-agent command line (everything after the binary
// name, --version excluded — that stays in cmd/eitri-agent). It parses flags,
// opens the state directory, then either enrolls the host ("join <blob>") or
// runs the agent.
func RunCLI(args []string) error {
cfg, rest, err := parseConfig(args)
if err != nil {
return err
}
st, err := state.Open(cfg.StateDir)
if err != nil {
return fmt.Errorf("open state dir: %w", err)
}
if len(rest) > 0 && rest[0] == "join" {
var blob string
if len(rest) > 1 {
blob = rest[1]
}
return join(st, cfg, blob)
}
return serve(st, cfg)
}
func parseConfig(args []string) (Config, []string, error) {
return parseConfigOn(runtime.GOOS, args)
}
// parseConfigOn defines and parses the agent flags, returning the assembled
// Config and any positional arguments (the "join <blob>" subcommand). Resource
// caps are validated here so a negative cap is rejected before any state is
// touched. goos decides whether --host-network is refused (see
// validateHostNetworks); production always passes runtime.GOOS, and it is a
// parameter — not a read of runtime.GOOS inline — so this refusal is provable
// on every platform a test runs on, not only the one the flag actually parses
// under. goos governs ONLY that one refusal: defaultStateDir and every other
// per-platform choice below still come from the build-tagged wire_*.go for
// this binary's real GOOS — passing "darwin" here does not simulate a Mac.
func parseConfigOn(goos string, args []string) (Config, []string, error) {
fs := flag.NewFlagSet("eitri-agent", flag.ContinueOnError)
// The default is per-platform: a Linux host keeps state where its
// root-run unit can create it, a Mac in a dotdir the running account owns.
// See each wire_*.go — importing one platform's filesystem layout into the
// other is how a Mac ended up being told to create /var/lib by hand.
stateDirDefault, err := defaultStateDir()
if err != nil {
return Config{}, nil, err
}
stateDir := fs.String("state-dir", stateDirDefault, "agent state directory")
// --ch-bin/--firmware/--bootstrap-url configure the cloud-hypervisor
// backend and --vfkit-bin the macOS one. They are accepted everywhere and
// ignored by platforms that don't run them, so the agent has one flag
// surface on every host.
chBin := fs.String("ch-bin", "cloud-hypervisor", "path to cloud-hypervisor binary")
firmware := fs.String("firmware", "/usr/share/eitri/CLOUDHV.fd", "path to CH UEFI firmware (CLOUDHV.fd)")
// Bare name by default: Homebrew installs to /opt/homebrew/bin on Apple
// Silicon and /usr/local/bin on Intel, so $PATH is the only answer that is
// right on both.
vfkitBin := fs.String("vfkit-bin", "vfkit", "path to the vfkit binary (macOS hosts)")
bridgeCIDR := fs.String("bridge-cidr", "", "the subnet this host's guests are on (Linux); empty adopts the control plane's suggestion at enrollment. Ignored where the host OS owns the guest network")
bootstrapURL := fs.String("bootstrap-url", "https://eitri.sh/dl/latest/manifest.json", "eitri.sh release manifest to fetch cloud-hypervisor/firmware from if missing at startup (empty disables bootstrap)")
tombstoneGrace := fs.Duration("tombstone-grace", 5*time.Minute, "quarantine period for tombstoned VMs before destroy")
vanishGrace := fs.Duration("vanish-grace", time.Hour, "quarantine period for VMs that vanished without a tombstone")
vmTimeout := fs.Duration("vm-timeout", 15*time.Minute, "watchdog bound on ONE VM's reconcile pass (0 disables); keep above the 10m image-download timeout")
imageCacheMaxGB := fs.Int64("image-cache-max-gb", 20, "evict least-recently-used cached base images beyond this size (0 = never evict)")
maxConcurrentCreates := fs.Int("max-concurrent-creates", 4, "cap how many VMs may be inside the I/O-heavy part of create at once (image fetch + disk copy); 0 = unlimited")
maxVCPUs := fs.Int64("max-vcpus", 0, "cap the total vCPUs this host offers the fleet (0 = unlimited; reserves headroom, advertised + enforced at boot)")
maxMemMB := fs.Int64("max-mem-mb", 0, "cap the total memory (MB) this host offers the fleet (0 = unlimited)")
maxDiskGB := fs.Int64("max-disk-gb", 0, "cap the total disk (GB) this host offers the fleet (0 = unlimited)")
hostNetworks := map[string]string{}
fs.Func("host-network", "a named guest network backed by a bridge this host's "+
"operator already declared, as name=bridge (repeatable; Linux hosts only — "+
"eitri attaches taps to the bridge, it never creates or addresses it)",
func(v string) error {
name, bridge, ok := strings.Cut(v, "=")
if !ok || bridge == "" {
return fmt.Errorf("want name=bridge, got %q", v)
}
if !names.IsNetworkName(name) {
return fmt.Errorf("invalid network name %q (1-32 of [a-z0-9-], no leading/trailing hyphen, \"nat\" reserved)", name)
}
if _, dup := hostNetworks[name]; dup {
return fmt.Errorf("network %q declared twice", name)
}
hostNetworks[name] = bridge
return nil
})
if err := fs.Parse(args); err != nil {
return Config{}, nil, err
}
if err := validateHostNetworks(names.OSServesNamedNetworks(goos), hostNetworks); err != nil {
return Config{}, nil, err
}
// Fixed order so the reported cap is deterministic when more than one is bad.
caps := []struct {
flag string
v int64
}{
{"max-vcpus", *maxVCPUs},
{"max-mem-mb", *maxMemMB},
{"max-disk-gb", *maxDiskGB},
}
for _, c := range caps {
if c.v < 0 {
return Config{}, nil, fmt.Errorf("resource cap --%s must be >= 0 (0 = unlimited), got %d", c.flag, c.v)
}
}
return Config{
StateDir: *stateDir,
CHBin: *chBin,
Firmware: *firmware,
VfkitBin: *vfkitBin,
BridgeCIDR: *bridgeCIDR,
BootstrapURL: *bootstrapURL,
TombstoneGrace: *tombstoneGrace,
VanishGrace: *vanishGrace,
VMTimeout: *vmTimeout,
ImageCacheMaxGB: *imageCacheMaxGB,
MaxConcurrentCreates: *maxConcurrentCreates,
MaxVCPUs: *maxVCPUs,
MaxMemMB: *maxMemMB,
MaxDiskGB: *maxDiskGB,
HostNetworks: hostNetworks,
}, fs.Args(), nil
}
// validateHostNetworks refuses --host-network on a host that cannot serve it.
// It takes the platform's support as a bool rather than a goos string so the
// one fact (names.OSServesNamedNetworks) has a single call site in this file,
// in parseConfigOn.
//
// Refusing here — at parse time, before state is opened or a bridge touched —
// matches this project's fail-loudly-at-boot precedent: a Mac that swallowed
// the flag would advertise no named networks anyway, and the operator's next
// signal would be a 409 telling them to add the exact flag they already
// passed. That is a dead end forever; this is a dead end at startup.
func validateHostNetworks(platformSupportsHostNetworks bool, hostNetworks map[string]string) error {
if len(hostNetworks) == 0 || platformSupportsHostNetworks {
return nil
}
return errors.New("--host-network: named networks are served by Linux hosts only; remove the flag (bridged guests need one of those)")
}
// join handles the "join <blob>" subcommand: decode the join blob, enroll,
// and persist identity — pinning the server cert from the blob (the enroll
// response's fingerprint is ignored, so the blob is the sole trust root).
func join(st *state.Store, cfg Config, blob string) error {
if blob == "" {
return errors.New("usage: eitri-agent join <join-blob>")
}
f, err := joinblob.Decode(blob)
if err != nil {
return fmt.Errorf("invalid join blob: %w", err)
}
hostname, err := os.Hostname()
if err != nil {
hostname = "unknown"
}
result, err := enrollclient.New(f.HTTPURL).Enroll(context.Background(), enrollclient.Request{
Token: f.Token,
Name: hostname,
OS: runtime.GOOS,
Arch: runtime.GOARCH,
Provisioner: platformProvisioner,
BridgeCIDR: proposeGuestCIDR(cfg),
})
if errors.Is(err, enrollclient.ErrTokenRejected) {
return errors.New("enroll rejected: token already used or expired — mint a new join token")
}
if err != nil {
return fmt.Errorf("enroll failed: %w", err)
}
// The response's bridge_cidr is a SUGGESTION, not a value to store: the
// resolver owns that field. Copying it here would reinstate server-wins on
// every re-enroll, which is the whole thing being removed.
id := state.Identity{
HostID: result.HostID,
Credential: result.Credential,
ServerQUICAddr: f.QUICAddr,
ServerCertSHA256: f.CertFP, // authoritative; response fingerprint ignored
}
if err := st.SaveIdentity(id); err != nil {
return fmt.Errorf("enroll succeeded but this host is NOT joined: nothing was written, and the blob's one-shot token is now spent, so a retry needs a freshly minted one: %w", err)
}
guestCIDR := resolveGuestCIDR(st, cfg, result.BridgeCIDR)
fmt.Printf("Enrolled: host_id=%s guest_cidr=%s\n", result.HostID, guestCIDR)
return nil
}
// serve handles the normal (no subcommand) run mode: it wires the reconcile
// engine and sync client and blocks until SIGINT/SIGTERM.
func serve(st *state.Store, cfg Config) error {
// First, before this agent touches the network, a bridge, or a VM record:
// claim the state directory. Everything below assumes this process is the
// only one driving this identity — the VM records it reconciles, the epoch
// fence it advances, and the one sync session the control plane keeps per
// host. Held until the process exits.
lk, err := statelock.Acquire(cfg.StateDir)
if err != nil {
return err
}
defer func() { _ = lk.Release() }()
id, ok := st.Identity()
if !ok {
return errors.New("not enrolled — run with 'join <blob>' subcommand first")
}
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer cancel()
// Flush integration-coverage counters on SIGUSR1 (no-op unless built with
// -cover and GOCOVERDIR is set). Lets the deploy boot-gate snapshot coverage
// from the live agent without bouncing the process.
covsnap.Install(ctx)
plat, err := newPlatform(ctx, cfg, st)
if err != nil {
return err
}
prov, pumps := plat.Prov, plat.Pumps
if recs, err := st.LoadVMs(); err == nil {
for _, rec := range recs {
if prov.Running(rec.Spec.VMID) {
pumps.Ensure(rec.Spec.VMID)
}
}
} else {
slog.Warn("state load failed; surviving VMs' consoles will be silent until reconcile restarts them", "err", err)
}
cache := imagecache.New(st.ImagesDir())
// Clamp before shifting: GB<<30 overflows int64 for absurd flag values —
// same trap class cloudhv's maxDiskGB comment documents. 1 PiB is beyond
// any real cache; anything above disables eviction just like 0 would.
imageCacheMaxGB := cfg.ImageCacheMaxGB
if imageCacheMaxGB < 0 || imageCacheMaxGB > 1<<20 {
slog.Warn("image-cache-max-gb out of range [0, 2^20]; disabling eviction", "value", imageCacheMaxGB)
imageCacheMaxGB = 0
}
cache.MaxBytes = imageCacheMaxGB << 30
// Reclaim temps left by a previous agent killed mid-fetch or mid-convert.
// Must run here, before the reconcile loop starts fetching: a sweep cannot
// tell an abandoned temp from one an in-flight fetch is still writing.
cache.SweepTemps()
engine := &reconcile.Engine{
St: st,
Prov: prov,
Images: cache.Ensure,
Seed: seed.Build,
HostKey: state.LoadOrCreateHostKey,
AgentVersion: version.Version,
BootID: hostinfo.BootID,
Now: time.Now,
TombstoneGrace: cfg.TombstoneGrace,
VanishGrace: cfg.VanishGrace,
MaxCreateAttempts: 3,
MaxConcurrentCreates: cfg.MaxConcurrentCreates,
VMTimeout: cfg.VMTimeout,
MaxVCPUs: cfg.MaxVCPUs,
MaxMemMB: cfg.MaxMemMB,
MaxDiskGB: cfg.MaxDiskGB,
}
// Seed the admission ledger from persisted records so the first reconcile
// accounts for VMs that survived the agent restart (their compute must
// count against the caps before any new create is admitted).
if recs, err := st.LoadVMs(); err == nil {
engine.SeedLedger(recs)
}
// The per-VM reconcile workers are deliberately NOT torn down here. Stopping
// the engine is terminal — it would make the next report an empty actual state,
// which the control plane reads as every VM on this host having vanished — and
// it would race an unjoined session worker that can still be mid-Step when ctx
// is cancelled (see the note below where pumps are torn down). The process is
// exiting; the OS reclaims the goroutines.
// The proxy publishing this host's guest ports. It is built here, in the
// composition root, because it needs one thing from the agent and nothing
// from any backend: where a guest is right now.
proxy := exposeproxy.NewManager(func(vmID string) string { return guestAddr(st, vmID) })
// The control plane's address, resolved once: the uplink address is asked on
// every report, and a per-tick resolution would put DNS on the report path,
// where a slow resolver stalls the session's writer. Only the route lookup
// uses this — the sync connection dials the configured address itself.
uplinkVia := id.ServerQUICAddr
if ua, err := net.ResolveUDPAddr("udp", uplinkVia); err == nil {
uplinkVia = ua.String()
} else {
slog.Warn("uplink target unresolved; per-report DNS lookups", "addr", uplinkVia, "err", err)
}
client := &syncclient.Client{
Engine: engine,
St: st,
Identity: id,
StateDir: cfg.StateDir,
Provisioner: platformProvisioner,
GuestCIDR: plat.GuestCIDR,
HostNetworks: plat.HostNetworks,
Runner: hostRunner,
Console: pumps,
Exposures: proxy,
UplinkAddr: func() string { return hostinfo.UplinkAddr(uplinkVia) },
MaxVCPUs: cfg.MaxVCPUs,
MaxMemMB: cfg.MaxMemMB,
MaxDiskGB: cfg.MaxDiskGB,
}
slog.Info("agent started", "host_id", id.HostID, "guest_cidr", plat.GuestCIDR())
client.Run(ctx)
// client.Run blocks until ctx is cancelled (SIGINT/SIGTERM) and only then
// returns, so the reconcile/sync loop is done driving VMs and no further
// Ensure/Stop calls are expected to reach pumps (an unjoined session
// worker could in principle still be mid-Engine.Step, but it has nothing
// left to drive once client.Run has returned). Tear down every serial
// console pump and every published listener here, at the very end of agent
// shutdown — a port this host no longer serves must not stay bound.
pumps.StopAll()
proxy.StopAll()
return nil
}