rollcall/cmd/rollcall/store.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

const schema = `
CREATE TABLE IF NOT EXISTS machines (
  id              text PRIMARY KEY,
  uuid            text,
  serial          text,
  hostname        text,
  manufacturer    text,
  product         text,
  state           text NOT NULL DEFAULT 'new',
  first_seen      timestamptz NOT NULL DEFAULT now(),
  last_seen       timestamptz,
  commissioned_at timestamptz,
  macs            jsonb NOT NULL DEFAULT '[]',
  bmc             jsonb,
  bmc_secret      bytea,
  inventory       jsonb
);
CREATE INDEX IF NOT EXISTS machines_uuid ON machines(uuid);
CREATE INDEX IF NOT EXISTS machines_macs ON machines USING gin(macs);
CREATE TABLE IF NOT EXISTS events (
  id         bigserial PRIMARY KEY,
  ts         timestamptz NOT NULL DEFAULT now(),
  machine_id text NOT NULL,
  event      text NOT NULL,
  detail     text
);
CREATE INDEX IF NOT EXISTS events_machine ON events(machine_id, ts);
CREATE TABLE IF NOT EXISTS profiles (
  name        text PRIMARY KEY,
  image       text NOT NULL,
  append      text,
  overlays    jsonb NOT NULL DEFAULT '[]',
  name_format text,
  vars        jsonb NOT NULL DEFAULT '{}',
  created     timestamptz NOT NULL DEFAULT now()
);
ALTER TABLE machines ADD COLUMN IF NOT EXISTS profile text REFERENCES profiles(name);
ALTER TABLE machines ADD COLUMN IF NOT EXISTS node_name text UNIQUE;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS mgmt_ip text;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS ib_ip text;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS node_token_hash text;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS node_token bytea;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS host_key bytea;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS booted_at timestamptz;
CREATE INDEX IF NOT EXISTS machines_node_token ON machines(node_token_hash);
ALTER TABLE machines ADD COLUMN IF NOT EXISTS os text;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS os_version text;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS deployed_at timestamptz;
CREATE TABLE IF NOT EXISTS jobs (
  id         text PRIMARY KEY,
  machine_id text NOT NULL,
  kind       text NOT NULL,
  state      text NOT NULL DEFAULT 'pending',
  spec       jsonb NOT NULL,
  progress   jsonb,
  token_hash text,
  token      bytea,
  created    timestamptz NOT NULL DEFAULT now(),
  started    timestamptz,
  finished   timestamptz,
  error      text
);
ALTER TABLE jobs ADD COLUMN IF NOT EXISTS bound_ip text;
CREATE INDEX IF NOT EXISTS jobs_machine ON jobs(machine_id, created);
CREATE INDEX IF NOT EXISTS jobs_token ON jobs(token_hash);
CREATE TABLE IF NOT EXISTS tags (
  name         text PRIMARY KEY,
  comment      text,
  match        text,
  cloud_config text,
  overlays     jsonb NOT NULL DEFAULT '[]',
  created      timestamptz NOT NULL DEFAULT now()
);
ALTER TABLE machines ADD COLUMN IF NOT EXISTS tags jsonb NOT NULL DEFAULT '[]';
ALTER TABLE machines ADD COLUMN IF NOT EXISTS auto_tags jsonb NOT NULL DEFAULT '[]';
CREATE INDEX IF NOT EXISTS machines_tags ON machines USING gin(tags);
CREATE INDEX IF NOT EXISTS machines_auto_tags ON machines USING gin(auto_tags);
ALTER TABLE machines ADD COLUMN IF NOT EXISTS building text;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS room text;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS rack text;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS rack_unit int;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS asset_tag text;
CREATE TABLE IF NOT EXISTS racks (
  building text NOT NULL,
  room     text NOT NULL,
  name     text NOT NULL,
  height   int  NOT NULL DEFAULT 42,
  comment  text,
  PRIMARY KEY (building, room, name)
);
ALTER TABLE machines ADD COLUMN IF NOT EXISTS rack_controller text;
ALTER TABLE machines ADD COLUMN IF NOT EXISTS net_layout text;
ALTER TABLE tags ADD COLUMN IF NOT EXISTS net_layout text;
CREATE TABLE IF NOT EXISTS fabrics (
  name    text PRIMARY KEY,
  kind    text NOT NULL DEFAULT 'ethernet',
  comment text
);
CREATE TABLE IF NOT EXISTS vlans (
  id             bigserial PRIMARY KEY,
  fabric         text NOT NULL REFERENCES fabrics(name) ON DELETE CASCADE,
  vid            int  NOT NULL DEFAULT 0,
  name           text,
  mtu            int  NOT NULL DEFAULT 1500,
  dhcp           boolean NOT NULL DEFAULT false,
  primary_rack   text,
  secondary_rack text,
  UNIQUE (fabric, vid)
);
CREATE TABLE IF NOT EXISTS subnets (
  cidr    text PRIMARY KEY,
  name    text,
  vlan_id bigint REFERENCES vlans(id) ON DELETE SET NULL,
  purpose text NOT NULL DEFAULT 'provisioning',
  gateway text,
  dns     jsonb NOT NULL DEFAULT '[]',
  domain  text,
  comment text
);
CREATE TABLE IF NOT EXISTS ip_ranges (
  id       bigserial PRIMARY KEY,
  subnet   text NOT NULL REFERENCES subnets(cidr) ON DELETE CASCADE,
  kind     text NOT NULL,
  start_ip text NOT NULL,
  end_ip   text NOT NULL,
  comment  text
);
CREATE TABLE IF NOT EXISTS ip_addresses (
  ip         text PRIMARY KEY,
  subnet     text NOT NULL REFERENCES subnets(cidr) ON DELETE CASCADE,
  machine_id text,
  iface      text,
  kind       text NOT NULL,
  hostname   text,
  created    timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS ip_addresses_machine ON ip_addresses(machine_id);
CREATE TABLE IF NOT EXISTS net_layouts (
  name    text PRIMARY KEY,
  spec    jsonb NOT NULL,
  comment text
);
CREATE TABLE IF NOT EXISTS rack_controllers (
  name            text PRIMARY KEY,
  url             text,
  kind            text NOT NULL DEFAULT 'remote',
  version         text,
  interfaces      jsonb NOT NULL DEFAULT '[]',
  dhcp_interfaces jsonb NOT NULL DEFAULT '[]',
  tftp_root       text,
  last_seen       timestamptz,
  registered      timestamptz NOT NULL DEFAULT now()
);
ALTER TABLE rack_controllers ADD COLUMN IF NOT EXISTS cert_serial text;
ALTER TABLE rack_controllers ADD COLUMN IF NOT EXISTS prev_cert_serial text;
ALTER TABLE rack_controllers ADD COLUMN IF NOT EXISTS prev_cert_until timestamptz;
ALTER TABLE rack_controllers ADD COLUMN IF NOT EXISTS cert_expires timestamptz;
ALTER TABLE rack_controllers ADD COLUMN IF NOT EXISTS arch text;
ALTER TABLE vlans ADD COLUMN IF NOT EXISTS dhcp_mode text NOT NULL DEFAULT '';
CREATE TABLE IF NOT EXISTS join_tokens (
  hash    text PRIMARY KEY,
  name    text,
  expires timestamptz NOT NULL,
  used    timestamptz,
  created timestamptz NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS pxe_files (
  mac        text PRIMARY KEY,
  machine_id text,
  rack       text,
  content    text NOT NULL,
  updated    timestamptz NOT NULL DEFAULT now()
);
`

var errNotFound = errors.New("not found")

type Store struct{ pool *pgxpool.Pool }

type Machine struct {
	ID             string          `json:"id"`
	UUID           *string         `json:"uuid"`
	Serial         *string         `json:"serial"`
	Hostname       *string         `json:"hostname"`
	Manufacturer   *string         `json:"manufacturer"`
	Product        *string         `json:"product"`
	State          string          `json:"state"`
	FirstSeen      time.Time       `json:"first_seen"`
	LastSeen       *time.Time      `json:"last_seen"`
	CommissionedAt *time.Time      `json:"commissioned_at"`
	MACs           []string        `json:"macs"`
	BMC            json.RawMessage `json:"bmc,omitempty"`
	Inventory      json.RawMessage `json:"inventory,omitempty"`
	Profile        *string         `json:"profile"`
	NodeName       *string         `json:"node_name"`
	MgmtIP         *string         `json:"mgmt_ip"`
	IBIP           *string         `json:"ib_ip"`
	BootedAt       *time.Time      `json:"booted_at"`
	OS             *string         `json:"os"`
	OSVersion      *string         `json:"os_version"`
	DeployedAt     *time.Time      `json:"deployed_at"`
	Tags           []string        `json:"tags"`      // manual
	AutoTags       []string        `json:"auto_tags"` // from tag match rules, recomputed at commissioning
	Location       Location        `json:"location"`
	RackController string          `json:"rack_controller"`
	NetLayout      string          `json:"net_layout"`
	bmcSecret      []byte
	nodeToken      []byte
	hostKey        []byte
}

type Event struct {
	TS     time.Time `json:"ts"`
	Event  string    `json:"event"`
	Detail string    `json:"detail"`
}

func openStore(ctx context.Context, url string) (*Store, error) {
	pool, err := pgxpool.New(ctx, url)
	if err != nil {
		return nil, err
	}
	if _, err := pool.Exec(ctx, schema); err != nil {
		return nil, fmt.Errorf("schema: %w", err)
	}
	if _, err := pool.Exec(ctx, adminCertSchema); err != nil {
		return nil, fmt.Errorf("schema: %w", err)
	}
	if _, err := pool.Exec(ctx, dnsRecordsSchema); err != nil {
		return nil, fmt.Errorf("schema: %w", err)
	}
	if _, err := pool.Exec(ctx, firmwareSchema); err != nil {
		return nil, fmt.Errorf("schema: %w", err)
	}
	return &Store{pool: pool}, nil
}

const machineCols = `id, uuid, serial, hostname, manufacturer, product, state, first_seen, last_seen,
	commissioned_at, macs, bmc, bmc_secret, inventory, profile, node_name, mgmt_ip, ib_ip, node_token, host_key, booted_at,
	os, os_version, deployed_at, tags, auto_tags,
	coalesce(building,''), coalesce(room,''), coalesce(rack,''), coalesce(rack_unit,0), coalesce(asset_tag,''),
	coalesce(rack_controller,''), coalesce(net_layout,'')`

func scanMachine(row pgx.Row, full bool) (*Machine, error) {
	m := &Machine{}
	var macs, tags, auto []byte
	err := row.Scan(&m.ID, &m.UUID, &m.Serial, &m.Hostname, &m.Manufacturer, &m.Product, &m.State,
		&m.FirstSeen, &m.LastSeen, &m.CommissionedAt, &macs, &m.BMC, &m.bmcSecret, &m.Inventory,
		&m.Profile, &m.NodeName, &m.MgmtIP, &m.IBIP, &m.nodeToken, &m.hostKey, &m.BootedAt,
		&m.OS, &m.OSVersion, &m.DeployedAt, &tags, &auto,
		&m.Location.Building, &m.Location.Room, &m.Location.Rack, &m.Location.Unit, &m.Location.AssetTag,
		&m.RackController, &m.NetLayout)
	if errors.Is(err, pgx.ErrNoRows) {
		return nil, errNotFound
	}
	if err != nil {
		return nil, err
	}
	_ = json.Unmarshal(macs, &m.MACs)
	_ = json.Unmarshal(tags, &m.Tags)
	_ = json.Unmarshal(auto, &m.AutoTags)
	if m.Tags == nil {
		m.Tags = []string{}
	}
	if m.AutoTags == nil {
		m.AutoTags = []string{}
	}
	if !full {
		m.Inventory = nil
	}
	return m, nil
}

func (s *Store) List(ctx context.Context) ([]*Machine, error) {
	rows, err := s.pool.Query(ctx, `SELECT `+machineCols+` FROM machines ORDER BY first_seen`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []*Machine
	for rows.Next() {
		m, err := scanMachine(rows, true)
		if err != nil {
			return nil, err
		}
		out = append(out, m)
	}
	if out == nil {
		out = []*Machine{}
	}
	return out, rows.Err()
}

func (s *Store) Get(ctx context.Context, id string) (*Machine, error) {
	return scanMachine(s.pool.QueryRow(ctx, `SELECT `+machineCols+` FROM machines WHERE id=$1`, id), true)
}

func (s *Store) Events(ctx context.Context, id string) ([]Event, error) {
	rows, err := s.pool.Query(ctx, `SELECT ts, event, coalesce(detail,'') FROM events WHERE machine_id=$1 ORDER BY ts`, id)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	evs := []Event{}
	for rows.Next() {
		var e Event
		if err := rows.Scan(&e.TS, &e.Event, &e.Detail); err != nil {
			return nil, err
		}
		evs = append(evs, e)
	}
	return evs, rows.Err()
}

type MachineEvent struct {
	Event
	MachineID string `json:"machine_id"`
}

// RecentEvents lists the newest events across all machines.
func (s *Store) RecentEvents(ctx context.Context, limit int) ([]MachineEvent, error) {
	rows, err := s.pool.Query(ctx, `SELECT ts, machine_id, event, coalesce(detail,'') FROM events ORDER BY ts DESC LIMIT $1`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	evs := []MachineEvent{}
	for rows.Next() {
		var e MachineEvent
		if err := rows.Scan(&e.TS, &e.MachineID, &e.Event.Event, &e.Detail); err != nil {
			return nil, err
		}
		evs = append(evs, e)
	}
	return evs, rows.Err()
}

func (s *Store) AddEvent(ctx context.Context, id, event, detail string) error {
	_, err := s.pool.Exec(ctx, `INSERT INTO events(machine_id, event, detail) VALUES ($1,$2,$3)`, id, event, detail)
	return err
}

// UpsertCommissioned stores a fresh inventory; returns true when the machine was new.
func (s *Store) UpsertCommissioned(ctx context.Context, m *Machine, secret []byte) (bool, error) {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return false, err
	}
	defer tx.Rollback(ctx)
	macs, _ := json.Marshal(m.MACs)
	tag, err := tx.Exec(ctx, `INSERT INTO machines(id, state) VALUES ($1,'new') ON CONFLICT (id) DO NOTHING`, m.ID)
	if err != nil {
		return false, err
	}
	isNew := tag.RowsAffected() == 1
	if isNew {
		if _, err := tx.Exec(ctx, `INSERT INTO events(machine_id, event) VALUES ($1,'enlisted')`, m.ID); err != nil {
			return false, err
		}
	}
	// a node that already has a profile goes back to 'stateless', everything else to 'ready'
	_, err = tx.Exec(ctx, `UPDATE machines SET uuid=$2, serial=$3, hostname=$4, manufacturer=$5, product=$6,
		state=CASE WHEN profile IS NULL THEN 'ready' ELSE 'stateless' END, booted_at=NULL, os=NULL, os_version=NULL, deployed_at=NULL,
		last_seen=now(), commissioned_at=now(), macs=$7, bmc=$8, bmc_secret=$9, inventory=$10 WHERE id=$1`,
		m.ID, m.UUID, m.Serial, m.Hostname, m.Manufacturer, m.Product, macs, m.BMC, secret, m.Inventory)
	if err != nil {
		return false, err
	}
	return isNew, tx.Commit(ctx)
}

func (s *Store) SetState(ctx context.Context, id, state string) error {
	tag, err := s.pool.Exec(ctx, `UPDATE machines SET state=$2 WHERE id=$1`, id, state)
	if err != nil {
		return err
	}
	if tag.RowsAffected() == 0 {
		return errNotFound
	}
	return nil
}

func (s *Store) Delete(ctx context.Context, id string) error {
	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx)
	if _, err := tx.Exec(ctx, `DELETE FROM events WHERE machine_id=$1`, id); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `DELETE FROM jobs WHERE machine_id=$1`, id); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `DELETE FROM ip_addresses WHERE machine_id=$1`, id); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `DELETE FROM pxe_files WHERE machine_id=$1`, id); err != nil {
		return err
	}
	tag, err := tx.Exec(ctx, `DELETE FROM machines WHERE id=$1`, id)
	if err != nil {
		return err
	}
	if tag.RowsAffected() == 0 {
		return errNotFound
	}
	return tx.Commit(ctx)
}