internal/server/store/exposures.go
Ref: Size: 7.8 KiB History
package store
import (
"database/sql"
"errors"
"fmt"
"time"
"github.com/a73x/eitri/internal/random"
sqlite "modernc.org/sqlite"
)
// Exposure is one published guest port: the fleet binds HostPort on the VM's
// host and pipes every accepted connection — or every datagram, when Protocol
// is "udp" — to GuestPort inside the guest. Tenant and HostID are both derived
// from the VM at create — never accepted from a caller — so an exposure can
// only ever name the partition and the machine its VM already lives in.
type Exposure struct {
ID, Tenant, VMID, HostID string
GuestPort, HostPort int64
Protocol, Scope string
CreatedAt time.Time
}
// The reserved host-port range. It is the host contract: one documented span
// that belongs to eitri on every host, so allocating a port needs no per-port
// hygiene reasoning anywhere else. A caller that names its own port is not
// held to it — only to the 1024 floor the API enforces.
const (
MinAllocatedHostPort = 30000
MaxAllocatedHostPort = 32767
)
// ErrExposureVMNotFound reports that the VM an exposure would belong to does
// not exist, or is tombstoned — a VM being torn down takes no new exposures.
var ErrExposureVMNotFound = errors.New("vm not found")
// ErrHostPortTaken reports that another exposure already holds the requested
// host port on that host, for the same protocol.
var ErrHostPortTaken = errors.New("host port already exposed on this host")
// ErrNoFreeHostPort reports that the reserved range is fully allocated on the
// host, so there is nothing left to hand out.
var ErrNoFreeHostPort = errors.New("no free host port in the reserved range")
// exposureColumns is the positional column list every exposure SELECT uses, so
// the order stays locked to scanExposure's positional Scan. Every column is
// table-qualified because one of the reads joins vms — the alias `e` is part
// of the contract each `from` clause below keeps.
const exposureColumns = `e.id, e.tenant, e.vm_id, e.host_id, e.guest_port, e.host_port, e.protocol, e.scope, e.created_at`
func scanExposure(rows *sql.Rows) (Exposure, error) {
var e Exposure
var createdAt string
if err := rows.Scan(&e.ID, &e.Tenant, &e.VMID, &e.HostID,
&e.GuestPort, &e.HostPort, &e.Protocol, &e.Scope, &createdAt); err != nil {
return Exposure{}, err
}
e.CreatedAt, _ = time.Parse(time.RFC3339, createdAt)
return e, nil
}
// queryExposures runs an exposureColumns-projected SELECT, with from appended
// verbatim after the column list (it carries the FROM clause, any join, the
// WHERE and the ORDER BY) and args bound in order.
func queryExposures(q querier, from string, args ...any) ([]Exposure, error) {
rows, err := q.Query(`SELECT `+exposureColumns+` `+from, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []Exposure
for rows.Next() {
e, err := scanExposure(rows)
if err != nil {
return nil, err
}
out = append(out, e)
}
return out, rows.Err()
}
// CreateExposure publishes guestPort of vmID on that VM's host, for protocol
// ("tcp" or "udp" — the API is what holds callers to those two). hostPort 0
// asks for one from the reserved range; a named port is honored or refused,
// and it is only taken when the SAME protocol already holds it: TCP 30000 and
// UDP 30000 are two ports on the host, and refusing the second would be the
// proxy inventing a scarcity the machine does not have. The whole decision —
// which host, which tenant, which port — happens in one transaction, so two
// concurrent creates cannot agree on the same port.
func (s *Store) CreateExposure(vmID string, guestPort, hostPort int64, protocol string) (Exposure, error) {
tx, err := s.db.Begin()
if err != nil {
return Exposure{}, err
}
defer tx.Rollback()
var tenant, hostID string
switch err := tx.QueryRow(
`SELECT tenant, host_id FROM vms WHERE id=? AND deleted_at IS NULL`, vmID,
).Scan(&tenant, &hostID); {
case errors.Is(err, sql.ErrNoRows):
return Exposure{}, ErrExposureVMNotFound
case err != nil:
return Exposure{}, fmt.Errorf("lookup vm: %w", err)
}
if hostPort == 0 {
hostPort, err = allocateHostPort(tx, hostID)
if err != nil {
return Exposure{}, err
}
}
e := Exposure{
ID: random.Hex(16), Tenant: tenant, VMID: vmID, HostID: hostID,
GuestPort: guestPort, HostPort: hostPort,
Protocol: protocol, Scope: "lan", CreatedAt: time.Now().UTC(),
}
if _, err := tx.Exec(
`INSERT INTO exposures(id, tenant, vm_id, host_id, guest_port, host_port, protocol, scope, created_at)
VALUES (?,?,?,?,?,?,?,?,?)`,
e.ID, e.Tenant, e.VMID, e.HostID, e.GuestPort, e.HostPort, e.Protocol, e.Scope,
e.CreatedAt.Format(time.RFC3339),
); err != nil {
// SQLITE_CONSTRAINT_UNIQUE (2067): the only UNIQUE constraint this
// insert can trip besides the random-hex PK is exposures_host_port_proto.
// Matched by errno, not message text, which embeds the index's column
// list and breaks on the next index change.
if serr, ok := errors.AsType[*sqlite.Error](err); ok && serr.Code() == 2067 {
return Exposure{}, ErrHostPortTaken
}
return Exposure{}, fmt.Errorf("insert exposure: %w", err)
}
if err := bumpEpoch(tx); err != nil {
return Exposure{}, fmt.Errorf("bump epoch: %w", err)
}
if err := tx.Commit(); err != nil {
return Exposure{}, err
}
return e, nil
}
// allocateHostPort returns the lowest port of the reserved range that no
// exposure on hostID holds — in EITHER protocol. A caller that named nothing
// gets a number it can hand to anyone without a protocol beside it, and the
// range is 2768 ports deep on a host that runs dozens of guests, so the ports
// this leaves on the table cost nothing that reasoning about half-taken
// numbers would not cost more. Naming a port explicitly still gets the full
// per-protocol answer.
//
// It reads the range's held ports into memory before choosing, because the
// store runs on a single connection: an open result set would block the insert
// that follows in the same transaction.
func allocateHostPort(tx *sql.Tx, hostID string) (int64, error) {
rows, err := tx.Query(
`SELECT host_port FROM exposures WHERE host_id=? AND host_port BETWEEN ? AND ?`,
hostID, MinAllocatedHostPort, MaxAllocatedHostPort)
if err != nil {
return 0, fmt.Errorf("read allocated host ports: %w", err)
}
defer rows.Close()
held := map[int64]bool{}
for rows.Next() {
var p int64
if err := rows.Scan(&p); err != nil {
return 0, err
}
held[p] = true
}
if err := rows.Err(); err != nil {
return 0, err
}
for p := int64(MinAllocatedHostPort); p <= MaxAllocatedHostPort; p++ {
if !held[p] {
return p, nil
}
}
return 0, ErrNoFreeHostPort
}
// DeleteExposure revokes an exposure. sql.ErrNoRows when there is none with
// that id. Bumps the epoch so the host's agent closes the listener.
func (s *Store) DeleteExposure(id string) error {
return s.mutate(`DELETE FROM exposures WHERE id=?`, id)
}
// GetExposure returns one exposure by id. sql.ErrNoRows when absent.
func (s *Store) GetExposure(id string) (Exposure, error) {
out, err := queryExposures(s.db, `FROM exposures e WHERE e.id=?`, id)
if err != nil {
return Exposure{}, err
}
if len(out) == 0 {
return Exposure{}, sql.ErrNoRows
}
return out[0], nil
}
// ListExposuresForVM returns one VM's exposures, lowest host port first.
func (s *Store) ListExposuresForVM(vmID string) ([]Exposure, error) {
return queryExposures(s.db, `FROM exposures e WHERE e.vm_id=? ORDER BY e.host_port`, vmID)
}
// ListExposuresForHost returns the exposures a host's agent should be running:
// its own, for VMs that are not tombstoned. A VM on its way out has already
// stopped being somewhere to send traffic, so its listener goes before its row
// does.
func (s *Store) ListExposuresForHost(hostID string) ([]Exposure, error) {
return queryExposures(s.db,
`FROM exposures e JOIN vms v ON v.id = e.vm_id
WHERE e.host_id=? AND v.deleted_at IS NULL ORDER BY e.host_port`, hostID)
}