internal/agent/state/state.go
Ref: Size: 14.2 KiB History
// Package state is the agent's durable state directory (default
// /var/lib/eitri-agent). Records are JSON, written atomically
// (tmp + rename) so a crash mid-write never corrupts a record.
package state
import (
"crypto/sha256"
"encoding/json"
"fmt"
"os"
"path/filepath"
"reflect"
"strconv"
"strings"
"time"
"github.com/a73x/eitri/internal/names"
)
type VMSpec struct {
VMID, Name, ImageURL, ImageSHA256, CloudInit, SSHAuthorizedKey string
// Network names the host network this guest gets a SECOND NIC on; "" is a
// guest with the NAT NIC alone. Additive, never instead: every guest is on
// the host's private bridge whatever this says. Part of the spec, not the
// record: it is desired state the control plane sent, and it must survive
// an agent restart so the replay can rebuild the VM's in-memory discovery
// state.
Network string
// VolumeIDs are the bound volumes to attach after root and seed, in this
// order — the first is /dev/vdc. Desired state that must survive an agent
// restart, like Network: a restarted agent re-attaches the same devices in
// the same order without waiting to be told again.
VolumeIDs []string
VCPUs, MemMB, DiskGB int64
}
// Equal is what == was before VolumeIDs made the struct uncomparable. Order is
// part of the answer: the first volume is /dev/vdc, so a reordered list is a
// different guest and reconcile treats it as an edit.
//
// DeepEqual rather than a hand-written field list, because the caller that
// matters is the one deciding whether a user edited a VM (reconcile.create,
// which hands an edited spec a fresh retry budget). A field added to this
// struct and forgotten in a field list would make edits to it invisible, and
// nothing would fail to say so. An empty list and no list are the same guest,
// so both normalise to nil first — s and o are copies, so this is local.
func (s VMSpec) Equal(o VMSpec) bool {
if len(s.VolumeIDs) == 0 {
s.VolumeIDs = nil
}
if len(o.VolumeIDs) == 0 {
o.VolumeIDs = nil
}
return reflect.DeepEqual(s, o)
}
// Disk is one block device attached to a VM, in attachment order. Index 0 is
// the root disk (/dev/vda) — both cloud-hypervisor and vfkit order by argument
// position and treat the first as root.
type Disk struct {
Path string
ReadOnly bool
}
type Record struct {
Spec VMSpec
// IP is the guest's address on the host's NAT underlay — allocated by this
// host, known before the guest boots, and what everything that has to reach
// the VM uses.
IP string
// NetworkIP is the address the site's DHCP server granted the guest on its
// named-network NIC, as this host snooped it. Empty for a guest with no
// such NIC, and until the first ACK crosses its tap. Durable so that an
// agent restart can report it again before the guest's next renewal is
// seen — the reservation table's twin for an address this host never gave.
NetworkIP string
BootID string // host boot ID at last start (lost-detection)
StopRequested bool // set BEFORE stopping: stopped != lost
QuarantinedAt *time.Time
QuarantineTombstoned bool // tombstoned (short grace) vs vanished (long grace)
CreateAttempts int // bounded retry before terminal failed (spec)
LastError string
CreatedAt time.Time
// HostPubKey is the public half of the guest's SSH host key, as an
// authorized_keys line. The private half is a file in this VM's directory
// and never leaves the host; this is what the agent reports upward so the
// control plane can certify it. Empty until the key has been generated.
HostPubKey string
}
type Identity struct {
HostID, Credential, BridgeCIDR string
ServerQUICAddr string // host:port for QUIC dial
ServerCertSHA256 string // pinned server cert fingerprint
}
type Store struct{ dir string }
// Open initialises the state directory, creating required subdirectories if
// they do not already exist. Returns a ready-to-use *Store.
func Open(dir string) (*Store, error) {
for _, sub := range []string{dir, filepath.Join(dir, "vms"), filepath.Join(dir, "images"), filepath.Join(dir, "volumes")} {
if err := os.MkdirAll(sub, 0o700); err != nil {
return nil, err
}
}
return &Store{dir: dir}, nil
}
// ImagesDir returns the path where downloaded images are cached.
func (s *Store) ImagesDir() string { return filepath.Join(s.dir, "images") }
// VolumesDir holds one directory per volume, BESIDE vms/ and never inside a
// VM's: a VM's directory is removed on destroy, a volume's is not. That
// separation is the durability.
func (s *Store) VolumesDir() string { return filepath.Join(s.dir, "volumes") }
// VolumeDir returns the per-volume directory for the given volume id.
func (s *Store) VolumeDir(id string) string { return filepath.Join(s.VolumesDir(), id) }
// VolumePath returns the path of the volume's backing file — the block device
// a guest is handed.
func (s *Store) VolumePath(id string) string { return filepath.Join(s.VolumeDir(id), "disk.raw") }
// VolumeTombstonePath is the marker whose mtime starts the reclaim grace. It
// is written once, on the first tick that sees the tombstone, and never
// refreshed: the grace runs from the deletion, not from the latest snapshot
// that restates it.
func (s *Store) VolumeTombstonePath(id string) string {
return filepath.Join(s.VolumeDir(id), "tombstoned")
}
// VMDir returns the per-VM directory for the given vmID.
func (s *Store) VMDir(vmID string) string { return filepath.Join(s.dir, "vms", vmID) }
// DiskPath returns the path of the VM's root disk image.
func (s *Store) DiskPath(vmID string) string { return filepath.Join(s.VMDir(vmID), "disk.raw") }
// SeedPath returns the path of the VM's cloud-init seed ISO.
func (s *Store) SeedPath(vmID string) string { return filepath.Join(s.VMDir(vmID), "seed.iso") }
// HostKeyPath returns the path of the VM's SSH host private key. It lives
// under the VM directory, so DeleteVM takes it with everything else.
func (s *Store) HostKeyPath(vmID string) string {
return filepath.Join(s.VMDir(vmID), "ssh_host_ed25519_key")
}
// SocketPath returns the path of the VM's cloud-hypervisor API socket.
func (s *Store) SocketPath(vmID string) string { return filepath.Join(s.VMDir(vmID), "ch.sock") }
// SerialSocketPath returns the path of the VM's cloud-hypervisor serial
// console socket (--serial socket=…). The serialpump dials it.
func (s *Store) SerialSocketPath(vmID string) string {
return filepath.Join(s.VMDir(vmID), "serial.sock")
}
// SerialLogPath returns the path of the VM's on-disk serial console log,
// written by serialpump so console history survives an agent restart.
func (s *Store) SerialLogPath(vmID string) string {
return filepath.Join(s.VMDir(vmID), "serial.log")
}
// MAC returns a deterministic, locally-administered MAC address for vmID.
// It uses the QEMU/KVM OUI prefix 52:54:00 and derives the last three octets
// from SHA-256(vmID). Shared by the hypervisor backend (guest NIC address) and
// the networking layer (DHCP reservation key) so both agree on one value
// without either importing the other.
func MAC(vmID string) string {
h := sha256.Sum256([]byte(vmID))
return fmt.Sprintf("52:54:00:%02x:%02x:%02x", h[0], h[1], h[2])
}
// NetMAC returns the deterministic MAC of the guest's SECOND NIC — the one
// attached to a named host network. Derived, like MAC(vmID), from a namespaced
// input ("net1:" + vmID) so a guest's two NICs can never collide and neither
// can two guests'. MAC(vmID) is left untouched by this: a VM's first NIC keeps
// the identity its DHCP reservation, its lease and its host's records were all
// filed under.
//
// It does not share MAC's OUI, on purpose. That NIC used to land on a private
// bridge this host alone spoke for; now it lands on the operator's real LAN,
// alongside whatever else is there — most of it, if it is QEMU or libvirt,
// answering under 52:54:00 too. The first octet here is 0x56: the
// locally-administered bit is set and the multicast bit is clear, so it is a
// legal unicast address, and because it differs from 0x52 no NetMAC can ever
// land in the same space as MAC's or as a neighbour's conventional guest MAC —
// not by any hash outcome, the two spaces are disjoint by construction. The
// remaining five octets carry SHA-256 of that input, instead of three,
// because a LAN is a bigger room than a private bridge: 40 bits of hash keeps
// a fleet's worth of guests comfortably below birthday-collision odds where
// 24 got uncomfortable at fleet scale.
//
// It is what the address snoop keys on, and what the guest's netplan matches
// the second stanza by, so both sides of a discovery agree without either
// asking the hypervisor what it ended up with.
func NetMAC(vmID string) string {
h := sha256.Sum256([]byte("net1:" + vmID))
return fmt.Sprintf("56:%02x:%02x:%02x:%02x:%02x", h[0], h[1], h[2], h[3], h[4])
}
// atomicWrite writes data to path using a tmp file + rename so that readers
// never see a partial write.
func atomicWrite(path string, data []byte) error {
return atomicWriteMode(path, data, 0o600)
}
// atomicWriteMode is atomicWrite with the resulting file's permissions stated
// outright. os.CreateTemp already opens 0600, but a file holding a private key
// should not owe its permissions to a helper's default.
func atomicWriteMode(path string, data []byte, perm os.FileMode) error {
dir := filepath.Dir(path)
f, err := os.CreateTemp(dir, ".tmp-")
if err != nil {
return err
}
tmpName := f.Name()
if err := f.Chmod(perm); err != nil {
f.Close()
os.Remove(tmpName)
return err
}
if _, err := f.Write(data); err != nil {
f.Close()
os.Remove(tmpName)
return err
}
if err := f.Close(); err != nil {
os.Remove(tmpName)
return err
}
return os.Rename(tmpName, path)
}
// SaveVM persists rec to disk atomically. The VM directory is created if needed.
func (s *Store) SaveVM(rec Record) error {
if err := os.MkdirAll(s.VMDir(rec.Spec.VMID), 0o700); err != nil {
return err
}
data, err := json.MarshalIndent(rec, "", " ")
if err != nil {
return err
}
return atomicWrite(filepath.Join(s.VMDir(rec.Spec.VMID), "record.json"), data)
}
// LoadVMs scans the vms/ subdirectory and returns all parseable records keyed
// by VMID. Unreadable or unparseable records are skipped — they indicate an
// incomplete create and the reconciler re-derives the correct state.
func (s *Store) LoadVMs() (map[string]Record, error) {
entries, err := os.ReadDir(filepath.Join(s.dir, "vms"))
if err != nil {
return nil, err
}
out := make(map[string]Record, len(entries))
for _, e := range entries {
if !e.IsDir() {
continue
}
recPath := filepath.Join(s.dir, "vms", e.Name(), "record.json")
raw, err := os.ReadFile(recPath)
if err != nil {
continue // incomplete create — skip
}
var rec Record
if err := json.Unmarshal(raw, &rec); err != nil {
continue // unparseable — skip
}
out[rec.Spec.VMID] = rec
}
return out, nil
}
// Get loads a single VM's record by id. ok reports that a record EXISTS; err is
// non-nil when a record exists but could not be read or parsed. Absence
// (ok=false, err=nil) is reported ONLY for a record that is genuinely not there.
//
// Callers MUST distinguish the two. "No record" means this host does not run
// this VM, so the agent's reconcile loop answers it by creating the VM — which
// rebuilds disk.raw and boots cloud-hypervisor. Letting an unreadable record
// read as absent would put a LIVE VM down that path: PrepareRootDisk over the disk
// its guest is running from, and a second hypervisor whose pidfile orphans the
// first. Anything that cannot be observed must be retried, never assumed empty.
//
// LoadVMs skips unreadable records instead, because a scan that cannot parse a
// file cannot say which VM it belonged to. A single-record read knows exactly
// which VM it failed to observe, so it says so.
func (s *Store) Get(vmID string) (Record, bool, error) {
raw, err := os.ReadFile(filepath.Join(s.VMDir(vmID), "record.json"))
if err != nil {
if os.IsNotExist(err) {
return Record{}, false, nil // no such record on this host
}
return Record{}, false, err
}
var rec Record
if err := json.Unmarshal(raw, &rec); err != nil {
return Record{}, false, err
}
return rec, true, nil
}
// DeleteVM removes the entire VM directory (record + disk + seed + socket).
func (s *Store) DeleteVM(vmID string) error {
return os.RemoveAll(s.VMDir(vmID))
}
// epochPath returns the path to the epoch file.
func (s *Store) epochPath() string { return filepath.Join(s.dir, "epoch") }
// Epoch returns the highest epoch this agent has seen, or 0 for a fresh
// store. DELIBERATE FAIL-OPEN: a corrupt epoch file also reads as 0, which
// resets the reaping fence; the quarantine grace period (not the fence) is
// the backstop in that case. SaveEpoch writes atomically, so corruption
// requires external interference, not a crash.
func (s *Store) Epoch() uint64 {
raw, err := os.ReadFile(s.epochPath())
if err != nil {
return 0
}
v, err := strconv.ParseUint(strings.TrimSpace(string(raw)), 10, 64)
if err != nil {
return 0
}
return v
}
// SaveEpoch writes epoch to disk atomically.
func (s *Store) SaveEpoch(epoch uint64) error {
return atomicWrite(s.epochPath(), []byte(strconv.FormatUint(epoch, 10)))
}
// identityPath returns the path to the identity file.
func (s *Store) identityPath() string { return filepath.Join(s.dir, "identity.json") }
// Identity reads the agent's persisted identity. Returns false if not yet enrolled.
func (s *Store) Identity() (Identity, bool) {
raw, err := os.ReadFile(s.identityPath())
if err != nil {
return Identity{}, false
}
var id Identity
if err := json.Unmarshal(raw, &id); err != nil {
return Identity{}, false
}
return id, true
}
// SaveIdentity persists the agent's identity atomically, refusing one whose
// pinned fingerprint the agent could never dial with.
func (s *Store) SaveIdentity(id Identity) error {
if !names.IsSHA256Hex(id.ServerCertSHA256) {
return fmt.Errorf("refusing to persist server cert fingerprint %q: it must be 64 lowercase hex, "+
"because it is the agent's only means of authenticating the control plane — "+
"an unusable one enrolls cleanly and then refuses every server this host ever dials",
id.ServerCertSHA256)
}
data, err := json.MarshalIndent(id, "", " ")
if err != nil {
return err
}
return atomicWrite(s.identityPath(), data)
}