clawbot
d84dd56460
check / check (push) Canceled after 0s
The live route tables lose their random UUID key; the row id the prefix-length sample draws is declared as the integer key. The prefix tables are keyed by prefix and keep only first_seen; a batch inserts only prefixes not stored yet. A route update sets only the columns that change. Each kept index names the queries using it. Every write goes through one connection kept for writing, with a 1 GiB page cache; reading connections get 16 MiB, heap limits rise to 1.5 and 2 GiB (README "Memory"). Batch writes go in key order. A benchmark run by hand writes 250,000 updates on 5 million routes faster than the feed's 10,000 a second; a live run follows. Model: opus-5-5
2661 lines
72 KiB
Go
2661 lines
72 KiB
Go
package database
|
|
|
|
import (
|
|
"cmp"
|
|
"context"
|
|
"database/sql"
|
|
_ "embed"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"maps"
|
|
"math/rand/v2"
|
|
"net"
|
|
"net/netip"
|
|
"os"
|
|
"path/filepath"
|
|
"runtime"
|
|
"slices"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.eeqj.de/sneak/routewatch/internal/config"
|
|
"git.eeqj.de/sneak/routewatch/internal/logger"
|
|
"git.eeqj.de/sneak/routewatch/pkg/asinfo"
|
|
_ "github.com/mattn/go-sqlite3" // CGO SQLite driver
|
|
)
|
|
|
|
// IMPORTANT: NO schema changes are to be made outside of schema.sql
|
|
// We do NOT support migrations. All schema changes MUST be made in schema.sql only.
|
|
//
|
|
//go:embed schema.sql
|
|
var dbSchema string
|
|
|
|
const (
|
|
dirPermissions = 0750 // rwxr-x---
|
|
ipVersionV4 = 4
|
|
ipVersionV6 = 6
|
|
ipv4Bits = 32
|
|
ipv6Bits = 128
|
|
|
|
tableLiveRoutesV4 = "live_routes_v4"
|
|
tableLiveRoutesV6 = "live_routes_v6"
|
|
|
|
// Names of the IP versions in error and log messages.
|
|
ipv4Family = "IPv4"
|
|
ipv6Family = "IPv6"
|
|
)
|
|
|
|
// The IP lookup does not use routes shorter than these mask lengths. A default
|
|
// route or a route this short only says where a peer sends traffic it has no
|
|
// better route for; it does not say who holds the address.
|
|
const (
|
|
minLookupMaskLengthV4 = 8
|
|
minLookupMaskLengthV6 = 16
|
|
)
|
|
|
|
// SQLite memory tuning (README "Memory"). cache_size and busy_timeout go in the
|
|
// DSN so every pooled connection gets them; the connection kept for writing
|
|
// then sets its own cache size. The heap limits are process-wide and set once.
|
|
const (
|
|
// maxConns is the size of the connection pool: the connection kept for
|
|
// writing and 9 reading connections.
|
|
maxConns = 10
|
|
// sqliteReadCacheSizeKiB is the page cache of each reading connection;
|
|
// negative means KiB. -16384 = 16 MiB, at most 144 MiB for the 9. A reading
|
|
// connection's cache is emptied whenever another connection has written
|
|
// since its last read, many times a minute, so a large one is wasted.
|
|
sqliteReadCacheSizeKiB = -16384
|
|
// sqliteWriteCacheSizeKiB is the page cache of the connection kept for
|
|
// writing, -1048576 = 1 GiB. It keeps its pages from one write to the next,
|
|
// so the index pages the route writes touch stay in memory.
|
|
sqliteWriteCacheSizeKiB = -1048576
|
|
// sqliteBusyTimeoutMs is how long a connection waits on a locked database.
|
|
sqliteBusyTimeoutMs = 5000
|
|
// sqliteSoftHeapLimitBytes (1.5 GiB) makes SQLite recycle its cache rather
|
|
// than allocate once its C heap passes this size. It is above the caches
|
|
// together, so it does not shrink them.
|
|
sqliteSoftHeapLimitBytes = 1610612736
|
|
// sqliteHardHeapLimitBytes (2 GiB) fails a statement with SQLITE_NOMEM
|
|
// instead of growing the C heap without bound.
|
|
sqliteHardHeapLimitBytes = 2147483648
|
|
)
|
|
|
|
// Settings of the ANALYZE that Analyze runs.
|
|
const (
|
|
// analysisLimit is about how many rows of each index ANALYZE reads
|
|
// (PRAGMA analysis_limit); SQLite estimates the rest.
|
|
analysisLimit = 1000
|
|
// optimizeAllTables is the PRAGMA optimize mask that runs ANALYZE (0x2)
|
|
// on every table that has an index without statistics or whose size has
|
|
// changed tenfold since it was last analyzed, not only on the tables this
|
|
// connection has queried (0x10000).
|
|
optimizeAllTables = 0x10002
|
|
)
|
|
|
|
// Bounds on the work GetRandomPrefixesByLength does to pick its sample.
|
|
const (
|
|
// maxPrefixesToShuffle is the most prefixes of one mask length that are
|
|
// read in full and shuffled to pick the sample. A mask length with more is
|
|
// sampled with random lookups.
|
|
maxPrefixesToShuffle = 5000
|
|
// sampleLookupsPerPrefix is the most random lookups made for each prefix
|
|
// asked for.
|
|
sampleLookupsPerPrefix = 4
|
|
)
|
|
|
|
// Common errors
|
|
var (
|
|
// ErrInvalidIP is returned when an IP address is malformed
|
|
ErrInvalidIP = errors.New("invalid IP address")
|
|
// ErrNoRoute is returned when no route is found for an IP
|
|
ErrNoRoute = errors.New("no route found")
|
|
// ErrNoStaleASN is returned when no ASN needs WHOIS refresh
|
|
ErrNoStaleASN = errors.New("no stale ASN found")
|
|
|
|
errInvalidASN = errors.New("invalid ASN")
|
|
errSameASNInPeering = errors.New("cannot create peering with same ASN")
|
|
)
|
|
|
|
// Database manages the SQLite database connection and operations.
|
|
type Database struct {
|
|
db *sql.DB
|
|
// writer is the connection kept for writing, taken from db when the
|
|
// database opens. Every write uses it while holding mu.
|
|
writer *sql.Conn
|
|
logger *logger.Logger
|
|
path string
|
|
mu sync.Mutex
|
|
lockedAt time.Time
|
|
lockedBy string
|
|
counts *liveCounts
|
|
}
|
|
|
|
// New creates a new database connection and initializes the schema.
|
|
func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
|
|
dbPath := filepath.Join(cfg.GetStateDir(), "db.sqlite")
|
|
|
|
// Log database path
|
|
logger.Info("Opening database", "path", dbPath)
|
|
|
|
// Ensure directory exists
|
|
dir := filepath.Dir(dbPath)
|
|
|
|
err := os.MkdirAll(dir, dirPermissions)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to create database directory: %w", err)
|
|
}
|
|
|
|
// Per-connection SQLite settings go in the DSN so every pooled connection
|
|
// gets them, not just the one that runs the Initialize pragmas. _txlock=
|
|
// immediate makes every transaction take the write lock at BEGIN. Without it
|
|
// a transaction that reads before writing starts as a reader and, when it
|
|
// then writes while another connection holds the write lock, fails at once
|
|
// with "database is locked" without waiting for _busy_timeout.
|
|
// _auto_vacuum=incremental lets Vacuum return free pages to the filesystem.
|
|
// SQLite only accepts it before the database file is first written, and the
|
|
// switch to WAL writes it, so it must be here: the driver applies it on open,
|
|
// before _journal_mode. On an existing file it changes nothing.
|
|
dsn := fmt.Sprintf(
|
|
"file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d"+
|
|
"&_auto_vacuum=incremental&_journal_mode=WAL&_txlock=immediate",
|
|
dbPath,
|
|
sqliteReadCacheSizeKiB,
|
|
sqliteBusyTimeoutMs,
|
|
)
|
|
|
|
db, err := sql.Open("sqlite3", dsn)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to open database: %w", err)
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
err = db.PingContext(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to ping database: %w", err)
|
|
}
|
|
|
|
// Set connection pool parameters
|
|
// Multiple connections allow concurrent reads while writes are serialized
|
|
db.SetMaxOpenConns(maxConns)
|
|
db.SetMaxIdleConns(maxConns)
|
|
db.SetConnMaxLifetime(0)
|
|
|
|
writer, err := db.Conn(ctx)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to open the writing connection: %w", err)
|
|
}
|
|
|
|
_, err = writer.ExecContext(ctx,
|
|
fmt.Sprintf("PRAGMA cache_size=%d", sqliteWriteCacheSizeKiB))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to set the writing connection's cache: %w",
|
|
err)
|
|
}
|
|
|
|
database := &Database{
|
|
db: db, writer: writer, logger: logger, path: dbPath,
|
|
counts: &liveCounts{},
|
|
}
|
|
|
|
err = database.Initialize()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to initialize database: %w", err)
|
|
}
|
|
|
|
// Seed the in-memory statistics counters from the tables once, before the
|
|
// streamer starts writing. From here on every write keeps them current, so
|
|
// the stats endpoints never scan the tables to report counts.
|
|
err = database.seedCounts(context.Background())
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to seed statistics counters: %w", err)
|
|
}
|
|
|
|
return database, nil
|
|
}
|
|
|
|
// Initialize creates the database schema if it doesn't exist.
|
|
func (d *Database) Initialize() error {
|
|
// Set SQLite pragmas for performance. Per-connection settings (cache_size,
|
|
// synchronous, busy_timeout, journal_mode) live in the DSN, and New sets the
|
|
// writing connection's cache_size; temp_store is left at its default so
|
|
// DISTINCT temp B-trees spill to disk instead of C heap. The heap limits
|
|
// below are process-wide, so setting them once here is enough for the whole
|
|
// pool.
|
|
pragmas := []string{
|
|
"PRAGMA journal_mode=WAL", // Write-Ahead Logging
|
|
fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes),
|
|
fmt.Sprintf("PRAGMA hard_heap_limit=%d", sqliteHardHeapLimitBytes),
|
|
}
|
|
|
|
for _, pragma := range pragmas {
|
|
err := d.exec(pragma)
|
|
if err != nil {
|
|
d.logger.Warn("Failed to set pragma", "pragma", pragma, "error", err)
|
|
}
|
|
}
|
|
|
|
// Run WAL checkpoint on startup to consolidate any existing WAL
|
|
var walPages, checkpointedPages, movedPages int
|
|
|
|
err := d.writer.QueryRowContext(context.Background(),
|
|
"PRAGMA wal_checkpoint(TRUNCATE)",
|
|
).Scan(&walPages, &checkpointedPages, &movedPages)
|
|
if err != nil {
|
|
d.logger.Warn("Failed to checkpoint WAL on startup", "error", err)
|
|
} else {
|
|
d.logger.Info("WAL checkpoint on startup",
|
|
"wal_pages", walPages,
|
|
"checkpointed", checkpointedPages,
|
|
"moved", movedPages,
|
|
)
|
|
}
|
|
|
|
err = d.exec(dbSchema)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Close closes the writing connection and then the connection pool.
|
|
func (d *Database) Close() error {
|
|
return errors.Join(d.writer.Close(), d.db.Close())
|
|
}
|
|
|
|
// A live-route upsert sets the origin AS of the peer's route for the prefix
|
|
// when the announcement changes it, then updates the rest of the route, and
|
|
// inserts the route only when that update matched no row. A live route is keyed
|
|
// by prefix and peer, so an announcement replaces the peer's route for the
|
|
// prefix whatever its origin AS. The update's rows-affected count (1 for an
|
|
// existing key, 0 for a new one) is what lets the in-memory route counters stay
|
|
// exact without a COUNT(*). Callers hold the database write lock, so no other
|
|
// writer can insert the same key between the statements.
|
|
//
|
|
// SQLite rewrites a row's entry in every index on a column an UPDATE sets, even
|
|
// when the value stays the same, so the update sets only the columns that
|
|
// change: mask_length follows from the prefix, and origin_asn has a statement
|
|
// of its own that changes nothing when the origin AS stays the same, as it
|
|
// mostly does.
|
|
const (
|
|
updateLiveRouteOriginV4SQL = `UPDATE live_routes_v4 SET origin_asn = ?
|
|
WHERE prefix = ? AND peer_ip = ? AND origin_asn != ?`
|
|
|
|
updateLiveRouteV4SQL = `UPDATE live_routes_v4
|
|
SET as_path = ?, next_hop = ?, last_updated = ?
|
|
WHERE prefix = ? AND peer_ip = ?`
|
|
|
|
insertLiveRouteV4SQL = `INSERT INTO live_routes_v4
|
|
(prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
|
|
last_updated) VALUES (?, ?, ?, ?, ?, ?, ?)`
|
|
|
|
updateLiveRouteOriginV6SQL = `UPDATE live_routes_v6 SET origin_asn = ?
|
|
WHERE prefix = ? AND peer_ip = ? AND origin_asn != ?`
|
|
|
|
updateLiveRouteV6SQL = `UPDATE live_routes_v6
|
|
SET as_path = ?, next_hop = ?, last_updated = ?
|
|
WHERE prefix = ? AND peer_ip = ?`
|
|
|
|
insertLiveRouteV6SQL = `INSERT INTO live_routes_v6
|
|
(prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
|
|
last_updated) VALUES (?, ?, ?, ?, ?, ?, ?)`
|
|
)
|
|
|
|
// Before a new route is inserted and after a route is deleted, the write looks
|
|
// up whether any live route has that prefix, to keep the in-memory prefix
|
|
// distribution exact. The lookup reads one entry of the prefix index.
|
|
const (
|
|
prefixHasLiveRouteV4SQL = `SELECT EXISTS
|
|
(SELECT 1 FROM live_routes_v4 WHERE prefix = ?)`
|
|
prefixHasLiveRouteV6SQL = `SELECT EXISTS
|
|
(SELECT 1 FROM live_routes_v6 WHERE prefix = ?)`
|
|
)
|
|
|
|
// A deletion removes the live route the peer has for the prefix, if any.
|
|
const (
|
|
deleteLiveRouteV4SQL = `DELETE FROM live_routes_v4
|
|
WHERE prefix = ? AND peer_ip = ?`
|
|
deleteLiveRouteV6SQL = `DELETE FROM live_routes_v6
|
|
WHERE prefix = ? AND peer_ip = ?`
|
|
)
|
|
|
|
// upsertRouteRow updates a live route with the origin and update statements of
|
|
// its IP version's table, inserting it when no row matched. It reports whether
|
|
// a new row was inserted and whether that row is the first live route for its
|
|
// prefix.
|
|
func upsertRouteRow(
|
|
ctx context.Context, setOrigin, upd, ins, has *sql.Stmt, route *LiveRoute,
|
|
pathJSON string,
|
|
) (bool, bool, error) {
|
|
_, err := setOrigin.ExecContext(ctx, route.OriginASN, route.Prefix,
|
|
route.PeerIP, route.OriginASN)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
|
|
res, err := upd.ExecContext(ctx, pathJSON, route.NextHop, route.LastUpdated,
|
|
route.Prefix, route.PeerIP)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
|
|
affected, err := res.RowsAffected()
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
|
|
if affected > 0 {
|
|
return false, false, nil
|
|
}
|
|
|
|
var prefixHadRoute bool
|
|
|
|
err = has.QueryRowContext(ctx, route.Prefix).Scan(&prefixHadRoute)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
|
|
_, err = ins.ExecContext(ctx, route.Prefix, route.MaskLength,
|
|
route.OriginASN, route.PeerIP, pathJSON, route.NextHop,
|
|
route.LastUpdated)
|
|
if err != nil {
|
|
return false, false, err
|
|
}
|
|
|
|
return true, !prefixHadRoute, nil
|
|
}
|
|
|
|
// upsertRoutesInTable upserts routes, all of one IP version, in the live route
|
|
// table of that version. It returns how many new rows it inserted and the mask
|
|
// lengths of the prefixes that got their first live route.
|
|
func upsertRoutesInTable(
|
|
ctx context.Context, tx *loggingTx, routes []*LiveRoute, ipVersion int,
|
|
) (int, []int, error) {
|
|
if len(routes) == 0 {
|
|
return 0, nil, nil
|
|
}
|
|
|
|
family, originSQL, updateSQL := ipv4Family, updateLiveRouteOriginV4SQL,
|
|
updateLiveRouteV4SQL
|
|
insertSQL, lookupSQL := insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
|
|
|
|
if ipVersion == ipVersionV6 {
|
|
family, originSQL, updateSQL = ipv6Family, updateLiveRouteOriginV6SQL,
|
|
updateLiveRouteV6SQL
|
|
insertSQL, lookupSQL = insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
|
|
}
|
|
|
|
setOrigin, err := tx.PrepareContext(ctx, originSQL)
|
|
if err != nil {
|
|
return 0, nil, fmt.Errorf("failed to prepare %s origin statement: %w",
|
|
family, err)
|
|
}
|
|
defer func() { _ = setOrigin.Close() }()
|
|
|
|
upd, err := tx.PrepareContext(ctx, updateSQL)
|
|
if err != nil {
|
|
return 0, nil, fmt.Errorf("failed to prepare %s update statement: %w",
|
|
family, err)
|
|
}
|
|
defer func() { _ = upd.Close() }()
|
|
|
|
ins, err := tx.PrepareContext(ctx, insertSQL)
|
|
if err != nil {
|
|
return 0, nil, fmt.Errorf("failed to prepare %s insert statement: %w",
|
|
family, err)
|
|
}
|
|
defer func() { _ = ins.Close() }()
|
|
|
|
has, err := tx.PrepareContext(ctx, lookupSQL)
|
|
if err != nil {
|
|
return 0, nil, fmt.Errorf(
|
|
"failed to prepare %s prefix lookup statement: %w", family, err)
|
|
}
|
|
defer func() { _ = has.Close() }()
|
|
|
|
var newRows int
|
|
// Mask lengths of the prefixes that get their first live route.
|
|
var newPrefixMaskLengths []int
|
|
|
|
for _, route := range routes {
|
|
pathJSON, err := json.Marshal(route.ASPath)
|
|
if err != nil {
|
|
return 0, nil, fmt.Errorf("failed to encode AS path: %w", err)
|
|
}
|
|
|
|
inserted, newPrefix, err := upsertRouteRow(ctx, setOrigin, upd, ins, has,
|
|
route, string(pathJSON))
|
|
if err != nil {
|
|
return 0, nil, fmt.Errorf("failed to upsert route %s: %w",
|
|
route.Prefix, err)
|
|
}
|
|
|
|
if inserted {
|
|
newRows++
|
|
}
|
|
|
|
if newPrefix {
|
|
newPrefixMaskLengths = append(newPrefixMaskLengths, route.MaskLength)
|
|
}
|
|
}
|
|
|
|
return newRows, newPrefixMaskLengths, nil
|
|
}
|
|
|
|
// UpsertLiveRouteBatch inserts or updates multiple live routes in a single
|
|
// transaction, the IPv4 routes first and then the IPv6 routes.
|
|
func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
|
|
if len(routes) == 0 {
|
|
return nil
|
|
}
|
|
|
|
d.lock("UpsertLiveRouteBatch")
|
|
defer d.unlock()
|
|
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to begin transaction: %w", err)
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
// The rows are written in key order, peer then prefix, so consecutive writes
|
|
// land on the same index pages. The sort is stable, so of two updates of
|
|
// one route the later one is still written last.
|
|
sorted := slices.SortedStableFunc(slices.Values(routes),
|
|
func(a, b *LiveRoute) int {
|
|
return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP),
|
|
cmp.Compare(a.Prefix, b.Prefix))
|
|
})
|
|
|
|
var routesV4, routesV6 []*LiveRoute
|
|
|
|
for _, route := range sorted {
|
|
if route.IPVersion == ipVersionV4 {
|
|
routesV4 = append(routesV4, route)
|
|
} else {
|
|
routesV6 = append(routesV6, route)
|
|
}
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
newV4, newPrefixMaskLengthsV4, err := upsertRoutesInTable(ctx, tx,
|
|
routesV4, ipVersionV4)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
newV6, newPrefixMaskLengthsV6, err := upsertRoutesInTable(ctx, tx,
|
|
routesV6, ipVersionV6)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to commit transaction: %w", err)
|
|
}
|
|
|
|
d.counts.addRoutes(newV4, newV6)
|
|
d.counts.addToDistribution(newPrefixMaskLengthsV4, newPrefixMaskLengthsV6, 1)
|
|
|
|
return nil
|
|
}
|
|
|
|
// deleteRouteRow runs one deletion with the delete statement of its IP
|
|
// version's table. It returns how many rows it removed and whether that left
|
|
// the prefix with no live route.
|
|
func deleteRouteRow(
|
|
ctx context.Context, deleteStmt, has *sql.Stmt, del LiveRouteDeletion,
|
|
) (int64, bool, error) {
|
|
res, err := deleteStmt.ExecContext(ctx, del.Prefix, del.PeerIP)
|
|
if err != nil {
|
|
return 0, false, fmt.Errorf("failed to delete route %s: %w",
|
|
del.Prefix, err)
|
|
}
|
|
|
|
// The feed often withdraws a route that is not live, which deletes
|
|
// nothing, so the in-memory route counters use the rows-affected count.
|
|
affected, err := res.RowsAffected()
|
|
if err != nil {
|
|
return 0, false, fmt.Errorf("failed to count deleted route %s: %w",
|
|
del.Prefix, err)
|
|
}
|
|
|
|
if affected == 0 {
|
|
return 0, false, nil
|
|
}
|
|
|
|
// The prefix leaves the distribution when no live route has it any more.
|
|
var prefixHasRoute bool
|
|
|
|
err = has.QueryRowContext(ctx, del.Prefix).Scan(&prefixHasRoute)
|
|
if err != nil {
|
|
return 0, false, fmt.Errorf("failed to look up prefix %s: %w",
|
|
del.Prefix, err)
|
|
}
|
|
|
|
return affected, !prefixHasRoute, nil
|
|
}
|
|
|
|
// deleteRoutesInTable runs deletions, all of one IP version, on the live route
|
|
// table of that version. It returns how many rows it removed and the mask
|
|
// lengths of the prefixes it left with no live route.
|
|
func deleteRoutesInTable(
|
|
ctx context.Context, tx *loggingTx, deletions []LiveRouteDeletion,
|
|
ipVersion int,
|
|
) (int64, []int, error) {
|
|
if len(deletions) == 0 {
|
|
return 0, nil, nil
|
|
}
|
|
|
|
family, deleteSQL, lookupSQL := ipv4Family, deleteLiveRouteV4SQL,
|
|
prefixHasLiveRouteV4SQL
|
|
if ipVersion == ipVersionV6 {
|
|
family, deleteSQL, lookupSQL = ipv6Family, deleteLiveRouteV6SQL,
|
|
prefixHasLiveRouteV6SQL
|
|
}
|
|
|
|
deleteStmt, err := tx.PrepareContext(ctx, deleteSQL)
|
|
if err != nil {
|
|
return 0, nil, fmt.Errorf("failed to prepare %s delete statement: %w",
|
|
family, err)
|
|
}
|
|
defer func() { _ = deleteStmt.Close() }()
|
|
|
|
has, err := tx.PrepareContext(ctx, lookupSQL)
|
|
if err != nil {
|
|
return 0, nil, fmt.Errorf(
|
|
"failed to prepare %s prefix lookup statement: %w", family, err)
|
|
}
|
|
defer func() { _ = has.Close() }()
|
|
|
|
var deleted int64
|
|
// Mask lengths of the prefixes left with no live route.
|
|
var gonePrefixMaskLengths []int
|
|
|
|
for _, del := range deletions {
|
|
affected, prefixGone, err := deleteRouteRow(ctx, deleteStmt, has, del)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
deleted += affected
|
|
|
|
if !prefixGone {
|
|
continue
|
|
}
|
|
|
|
maskLength, err := prefixMaskLength(del.Prefix)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
gonePrefixMaskLengths = append(gonePrefixMaskLengths, maskLength)
|
|
}
|
|
|
|
return deleted, gonePrefixMaskLengths, nil
|
|
}
|
|
|
|
// DeleteLiveRouteBatch deletes multiple live routes in a single transaction,
|
|
// the IPv4 routes first and then the IPv6 routes.
|
|
func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
|
|
if len(deletions) == 0 {
|
|
return nil
|
|
}
|
|
|
|
d.lock("DeleteLiveRouteBatch")
|
|
defer d.unlock()
|
|
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to begin transaction: %w", err)
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
// The rows are deleted in key order, peer then prefix, as
|
|
// UpsertLiveRouteBatch writes them.
|
|
sorted := slices.SortedFunc(slices.Values(deletions),
|
|
func(a, b LiveRouteDeletion) int {
|
|
return cmp.Or(cmp.Compare(a.PeerIP, b.PeerIP),
|
|
cmp.Compare(a.Prefix, b.Prefix))
|
|
})
|
|
|
|
var deletionsV4, deletionsV6 []LiveRouteDeletion
|
|
|
|
for _, del := range sorted {
|
|
if del.IPVersion == ipVersionV4 {
|
|
deletionsV4 = append(deletionsV4, del)
|
|
} else {
|
|
deletionsV6 = append(deletionsV6, del)
|
|
}
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
deletedV4, gonePrefixMaskLengthsV4, err := deleteRoutesInTable(ctx, tx,
|
|
deletionsV4, ipVersionV4)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
deletedV6, gonePrefixMaskLengthsV6, err := deleteRoutesInTable(ctx, tx,
|
|
deletionsV6, ipVersionV6)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to commit transaction: %w", err)
|
|
}
|
|
|
|
d.counts.addRoutes(-int(deletedV4), -int(deletedV6))
|
|
d.counts.addToDistribution(gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6, -1)
|
|
|
|
return nil
|
|
}
|
|
|
|
// Deleting all live routes of a peer. The first query of each table selects the
|
|
// mask length of each of the peer's routes whose prefix no other peer has a
|
|
// live route for: the delete takes those prefixes out of the prefix
|
|
// distribution.
|
|
const (
|
|
peerOnlyRouteMaskLengthsV4SQL = `SELECT r.mask_length
|
|
FROM live_routes_v4 AS r WHERE r.peer_ip = ?
|
|
AND NOT EXISTS (SELECT 1 FROM live_routes_v4 AS o
|
|
WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)`
|
|
deletePeerLiveRoutesV4SQL = `DELETE FROM live_routes_v4 WHERE peer_ip = ?`
|
|
|
|
peerOnlyRouteMaskLengthsV6SQL = `SELECT r.mask_length
|
|
FROM live_routes_v6 AS r WHERE r.peer_ip = ?
|
|
AND NOT EXISTS (SELECT 1 FROM live_routes_v6 AS o
|
|
WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)`
|
|
deletePeerLiveRoutesV6SQL = `DELETE FROM live_routes_v6 WHERE peer_ip = ?`
|
|
)
|
|
|
|
// deletePeerRoutes deletes all live routes of peerIP from one table, given that
|
|
// table's two statements above. It returns how many routes it deleted and the
|
|
// mask length of each prefix it left with no live route.
|
|
func deletePeerRoutes(
|
|
ctx context.Context, tx *loggingTx,
|
|
onlyRouteMaskLengthsSQL, deleteSQL, peerIP string,
|
|
) (int64, []int, error) {
|
|
rows, err := tx.QueryContext(ctx, onlyRouteMaskLengthsSQL, peerIP)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
var goneMaskLengths []int
|
|
|
|
for rows.Next() {
|
|
var maskLength int
|
|
|
|
err := rows.Scan(&maskLength)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
goneMaskLengths = append(goneMaskLengths, maskLength)
|
|
}
|
|
|
|
err = rows.Err()
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
res, err := tx.ExecContext(ctx, deleteSQL, peerIP)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
deleted, err := res.RowsAffected()
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
return deleted, goneMaskLengths, nil
|
|
}
|
|
|
|
// DeleteLiveRoutesOfPeer deletes every live route of the peer, IPv4 and IPv6,
|
|
// in one transaction, as when the peer's BGP session goes down. It returns how
|
|
// many routes it deleted.
|
|
func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
|
|
d.lock("DeleteLiveRoutesOfPeer")
|
|
defer d.unlock()
|
|
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to begin transaction: %w", err)
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
ctx := context.Background()
|
|
|
|
deletedV4, goneMaskLengthsV4, err := deletePeerRoutes(ctx, tx,
|
|
peerOnlyRouteMaskLengthsV4SQL, deletePeerLiveRoutesV4SQL, peerIP)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to delete the IPv4 routes of peer %s: %w",
|
|
peerIP, err)
|
|
}
|
|
|
|
deletedV6, goneMaskLengthsV6, err := deletePeerRoutes(ctx, tx,
|
|
peerOnlyRouteMaskLengthsV6SQL, deletePeerLiveRoutesV6SQL, peerIP)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to delete the IPv6 routes of peer %s: %w",
|
|
peerIP, err)
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to commit transaction: %w", err)
|
|
}
|
|
|
|
d.counts.addRoutes(-int(deletedV4), -int(deletedV6))
|
|
d.counts.addToDistribution(goneMaskLengthsV4, goneMaskLengthsV6, -1)
|
|
|
|
return int(deletedV4 + deletedV6), nil
|
|
}
|
|
|
|
// insertPrefixesInTable inserts the prefixes, all of one IP version, that the
|
|
// prefix table of that version does not hold yet, each first seen at its time
|
|
// in prefixes. A prefix already stored is left as it is. It returns how many it
|
|
// inserted.
|
|
func insertPrefixesInTable(
|
|
ctx context.Context, tx *loggingTx, prefixes map[string]time.Time,
|
|
ipVersion int,
|
|
) (int, error) {
|
|
if len(prefixes) == 0 {
|
|
return 0, nil
|
|
}
|
|
|
|
family := ipv4Family
|
|
insertSQL := "INSERT INTO prefixes_v4 (prefix, first_seen) VALUES (?, ?) " +
|
|
"ON CONFLICT (prefix) DO NOTHING"
|
|
|
|
if ipVersion == ipVersionV6 {
|
|
family = ipv6Family
|
|
insertSQL = "INSERT INTO prefixes_v6 (prefix, first_seen) VALUES (?, ?) " +
|
|
"ON CONFLICT (prefix) DO NOTHING"
|
|
}
|
|
|
|
insertStmt, err := tx.PrepareContext(ctx, insertSQL)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to prepare %s insert statement: %w",
|
|
family, err)
|
|
}
|
|
defer func() { _ = insertStmt.Close() }()
|
|
|
|
var newPrefixes int
|
|
|
|
// The rows are written in key order, as UpsertLiveRouteBatch writes routes.
|
|
for _, prefix := range slices.Sorted(maps.Keys(prefixes)) {
|
|
res, err := insertStmt.ExecContext(ctx, prefix, prefixes[prefix])
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to insert prefix %s: %w", prefix, err)
|
|
}
|
|
|
|
// One row for a new prefix, none for one already stored.
|
|
inserted, err := res.RowsAffected()
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to count inserted prefix %s: %w",
|
|
prefix, err)
|
|
}
|
|
|
|
if inserted > 0 {
|
|
newPrefixes++
|
|
}
|
|
}
|
|
|
|
return newPrefixes, nil
|
|
}
|
|
|
|
// UpdatePrefixesBatch stores, in a single transaction, the prefixes that are
|
|
// not stored yet, each first seen at its time in prefixes: the IPv4 prefixes
|
|
// first and then the IPv6 prefixes.
|
|
func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
|
|
if len(prefixes) == 0 {
|
|
return nil
|
|
}
|
|
|
|
d.lock("UpdatePrefixesBatch")
|
|
defer d.unlock()
|
|
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to begin transaction: %w", err)
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
prefixesV4 := make(map[string]time.Time)
|
|
prefixesV6 := make(map[string]time.Time)
|
|
|
|
for prefix, timestamp := range prefixes {
|
|
if detectIPVersion(prefix) == ipVersionV4 {
|
|
prefixesV4[prefix] = timestamp
|
|
} else {
|
|
prefixesV6[prefix] = timestamp
|
|
}
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
newV4, err := insertPrefixesInTable(ctx, tx, prefixesV4, ipVersionV4)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
newV6, err := insertPrefixesInTable(ctx, tx, prefixesV6, ipVersionV6)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to commit transaction: %w", err)
|
|
}
|
|
|
|
d.counts.addPrefixes(newV4, newV6)
|
|
|
|
return nil
|
|
}
|
|
|
|
// newASN returns a new ASN record first and last seen at timestamp, with the
|
|
// handle and description from the AS info lookup when it knows the number.
|
|
func newASN(number int, timestamp time.Time) ASN {
|
|
asn := ASN{
|
|
ASN: number,
|
|
FirstSeen: timestamp,
|
|
LastSeen: timestamp,
|
|
}
|
|
|
|
// Look up ASN info
|
|
if info, ok := asinfo.Get(number); ok {
|
|
asn.Handle = info.Handle
|
|
asn.Description = info.Description
|
|
}
|
|
|
|
return asn
|
|
}
|
|
|
|
// upsertASN updates the last_seen time of ASN number with the given
|
|
// statements, inserting the ASN when it is new. It reports whether it inserted
|
|
// it.
|
|
func upsertASN(
|
|
ctx context.Context, selectStmt, updateStmt, insertStmt *sql.Stmt,
|
|
number int, timestamp time.Time,
|
|
) (bool, error) {
|
|
var (
|
|
asn ASN
|
|
handle, description sql.NullString
|
|
)
|
|
|
|
err := selectStmt.QueryRowContext(ctx, number).Scan(&asn.ASN, &handle,
|
|
&description, &asn.FirstSeen, &asn.LastSeen)
|
|
if err == nil {
|
|
// ASN exists, update last_seen
|
|
_, err = updateStmt.ExecContext(ctx, timestamp, number)
|
|
if err != nil {
|
|
return false, fmt.Errorf("failed to update ASN %d: %w", number, err)
|
|
}
|
|
|
|
return false, nil
|
|
}
|
|
|
|
if !errors.Is(err, sql.ErrNoRows) {
|
|
return false, fmt.Errorf("failed to query ASN %d: %w", number, err)
|
|
}
|
|
|
|
// ASN doesn't exist, create it
|
|
asn = newASN(number, timestamp)
|
|
|
|
_, err = insertStmt.ExecContext(ctx, asn.ASN, asn.Handle,
|
|
asn.Description, asn.FirstSeen, asn.LastSeen)
|
|
if err != nil {
|
|
return false, fmt.Errorf("failed to insert ASN %d: %w", number, err)
|
|
}
|
|
|
|
return true, nil
|
|
}
|
|
|
|
// GetOrCreateASNBatch creates or updates multiple ASNs in a single transaction
|
|
func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
|
|
if len(asns) == 0 {
|
|
return nil
|
|
}
|
|
|
|
d.lock("GetOrCreateASNBatch")
|
|
defer d.unlock()
|
|
|
|
ctx := context.Background()
|
|
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to begin transaction: %w", err)
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
// Prepare statements
|
|
selectStmt, err := tx.PrepareContext(ctx,
|
|
"SELECT asn, handle, description, first_seen, last_seen "+
|
|
"FROM asns WHERE asn = ?")
|
|
if err != nil {
|
|
return fmt.Errorf("failed to prepare select statement: %w", err)
|
|
}
|
|
defer func() { _ = selectStmt.Close() }()
|
|
|
|
updateStmt, err := tx.PrepareContext(ctx,
|
|
"UPDATE asns SET last_seen = ? WHERE asn = ?")
|
|
if err != nil {
|
|
return fmt.Errorf("failed to prepare update statement: %w", err)
|
|
}
|
|
defer func() { _ = updateStmt.Close() }()
|
|
|
|
insertStmt, err := tx.PrepareContext(ctx,
|
|
"INSERT INTO asns (asn, handle, description, first_seen, last_seen) "+
|
|
"VALUES (?, ?, ?, ?, ?)")
|
|
if err != nil {
|
|
return fmt.Errorf("failed to prepare insert statement: %w", err)
|
|
}
|
|
defer func() { _ = insertStmt.Close() }()
|
|
|
|
var newASNs int
|
|
|
|
for number, timestamp := range asns {
|
|
inserted, err := upsertASN(ctx, selectStmt, updateStmt, insertStmt,
|
|
number, timestamp)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if inserted {
|
|
newASNs++
|
|
}
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to commit transaction: %w", err)
|
|
}
|
|
|
|
d.counts.addASNs(newASNs)
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetOrCreateASN retrieves an existing ASN or creates a new one if it doesn't
|
|
// exist.
|
|
func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error) {
|
|
d.lock("GetOrCreateASN")
|
|
defer d.unlock()
|
|
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
var (
|
|
asn ASN
|
|
handle, description sql.NullString
|
|
)
|
|
|
|
err = tx.QueryRow("SELECT asn, handle, description, first_seen, last_seen "+
|
|
"FROM asns WHERE asn = ?", number).
|
|
Scan(&asn.ASN, &handle, &description, &asn.FirstSeen, &asn.LastSeen)
|
|
if err == nil {
|
|
// ASN exists, update last_seen
|
|
asn.Handle = handle.String
|
|
asn.Description = description.String
|
|
|
|
_, err = tx.Exec("UPDATE asns SET last_seen = ? WHERE asn = ?",
|
|
timestamp, number)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
asn.LastSeen = timestamp
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
d.logger.Error("Failed to commit transaction for ASN update",
|
|
"asn", number, "error", err)
|
|
|
|
return nil, err
|
|
}
|
|
|
|
return &asn, nil
|
|
}
|
|
|
|
if !errors.Is(err, sql.ErrNoRows) {
|
|
return nil, err
|
|
}
|
|
|
|
// ASN doesn't exist, create it with ASN info lookup
|
|
asn = newASN(number, timestamp)
|
|
|
|
_, err = tx.Exec("INSERT INTO asns "+
|
|
"(asn, handle, description, first_seen, last_seen) "+
|
|
"VALUES (?, ?, ?, ?, ?)",
|
|
asn.ASN, asn.Handle, asn.Description, asn.FirstSeen, asn.LastSeen)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
d.logger.Error("Failed to commit transaction for ASN creation",
|
|
"asn", number, "error", err)
|
|
|
|
return nil, err
|
|
}
|
|
|
|
d.counts.addASNs(1)
|
|
|
|
return &asn, nil
|
|
}
|
|
|
|
// GetOrCreatePrefix retrieves an existing prefix or creates a new one if it
|
|
// doesn't exist.
|
|
func (d *Database) GetOrCreatePrefix(
|
|
prefix string, timestamp time.Time) (*Prefix, error) {
|
|
d.lock("GetOrCreatePrefix")
|
|
defer d.unlock()
|
|
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
// Determine table based on IP version
|
|
ipVersion := detectIPVersion(prefix)
|
|
|
|
tableName := "prefixes_v4"
|
|
if ipVersion == ipVersionV6 {
|
|
tableName = "prefixes_v6"
|
|
}
|
|
|
|
var p Prefix
|
|
|
|
query := fmt.Sprintf("SELECT prefix, first_seen FROM %s WHERE prefix = ?",
|
|
tableName)
|
|
|
|
err = tx.QueryRow(query, prefix).Scan(&p.Prefix, &p.FirstSeen)
|
|
if err == nil {
|
|
// Prefix exists; the deferred rollback ends the read.
|
|
p.IPVersion = ipVersion
|
|
|
|
return &p, nil
|
|
}
|
|
|
|
if !errors.Is(err, sql.ErrNoRows) {
|
|
return nil, err
|
|
}
|
|
|
|
// Prefix doesn't exist, create it
|
|
p = Prefix{Prefix: prefix, IPVersion: ipVersion, FirstSeen: timestamp}
|
|
insertQuery := fmt.Sprintf("INSERT INTO %s "+
|
|
"(prefix, first_seen) VALUES (?, ?)", tableName)
|
|
|
|
_, err = tx.Exec(insertQuery, p.Prefix, p.FirstSeen)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
d.logger.Error("Failed to commit transaction for prefix creation",
|
|
"prefix", prefix, "error", err)
|
|
|
|
return nil, err
|
|
}
|
|
|
|
if ipVersion == ipVersionV4 {
|
|
d.counts.addPrefixes(1, 0)
|
|
} else {
|
|
d.counts.addPrefixes(0, 1)
|
|
}
|
|
|
|
return &p, nil
|
|
}
|
|
|
|
// RecordAnnouncement inserts a new BGP announcement or withdrawal into the
|
|
// database.
|
|
func (d *Database) RecordAnnouncement(announcement *Announcement) error {
|
|
d.lock("RecordAnnouncement")
|
|
defer d.unlock()
|
|
|
|
err := d.exec(`
|
|
INSERT INTO announcements (id, prefix_id, peer_asn, origin_asn, path,
|
|
next_hop, timestamp, is_withdrawal)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
announcement.ID.String(), announcement.PrefixID.String(),
|
|
announcement.PeerASN, announcement.OriginASN,
|
|
announcement.Path, announcement.NextHop, announcement.Timestamp,
|
|
announcement.IsWithdrawal)
|
|
|
|
return err
|
|
}
|
|
|
|
// RecordPeering records a peering relationship between two ASNs.
|
|
func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error {
|
|
// Validate ASNs
|
|
if asA <= 0 || asB <= 0 {
|
|
return fmt.Errorf("%w: asA=%d, asB=%d", errInvalidASN, asA, asB)
|
|
}
|
|
|
|
if asA == asB {
|
|
return fmt.Errorf("%w: %d", errSameASNInPeering, asA)
|
|
}
|
|
|
|
// Normalize: ensure asA < asB
|
|
if asA > asB {
|
|
asA, asB = asB, asA
|
|
}
|
|
|
|
d.lock("RecordPeering")
|
|
defer d.unlock()
|
|
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
var exists bool
|
|
|
|
err = tx.QueryRow(
|
|
"SELECT EXISTS(SELECT 1 FROM peerings WHERE as_a = ? AND as_b = ?)",
|
|
asA, asB).Scan(&exists)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if exists {
|
|
_, err = tx.Exec(
|
|
"UPDATE peerings SET last_seen = ? WHERE as_a = ? AND as_b = ?",
|
|
timestamp, asA, asB)
|
|
} else {
|
|
_, err = tx.Exec(`
|
|
INSERT INTO peerings (id, as_a, as_b, first_seen, last_seen)
|
|
VALUES (?, ?, ?, ?, ?)`,
|
|
generateUUID().String(), asA, asB, timestamp, timestamp)
|
|
}
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
d.logger.Error("Failed to commit transaction for peering",
|
|
"as_a", asA,
|
|
"as_b", asB,
|
|
"error", err,
|
|
)
|
|
|
|
return err
|
|
}
|
|
|
|
if !exists {
|
|
d.counts.addPeerings(1)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// UpdatePeerBatch updates or creates multiple BGP peer records in a single transaction
|
|
func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error {
|
|
if len(peers) == 0 {
|
|
return nil
|
|
}
|
|
|
|
d.lock("UpdatePeerBatch")
|
|
defer d.unlock()
|
|
|
|
ctx := context.Background()
|
|
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to begin transaction: %w", err)
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
// Prepare statements
|
|
checkStmt, err := tx.PrepareContext(ctx,
|
|
"SELECT EXISTS(SELECT 1 FROM bgp_peers WHERE peer_ip = ?)")
|
|
if err != nil {
|
|
return fmt.Errorf("failed to prepare check statement: %w", err)
|
|
}
|
|
defer func() { _ = checkStmt.Close() }()
|
|
|
|
updateStmt, err := tx.PrepareContext(ctx,
|
|
"UPDATE bgp_peers SET peer_asn = ?, last_seen = ?, "+
|
|
"last_message_type = ? WHERE peer_ip = ?")
|
|
if err != nil {
|
|
return fmt.Errorf("failed to prepare update statement: %w", err)
|
|
}
|
|
defer func() { _ = updateStmt.Close() }()
|
|
|
|
insertStmt, err := tx.PrepareContext(ctx,
|
|
"INSERT INTO bgp_peers (id, peer_ip, peer_asn, first_seen, "+
|
|
"last_seen, last_message_type) VALUES (?, ?, ?, ?, ?, ?)")
|
|
if err != nil {
|
|
return fmt.Errorf("failed to prepare insert statement: %w", err)
|
|
}
|
|
defer func() { _ = insertStmt.Close() }()
|
|
|
|
var newPeers int
|
|
|
|
for _, update := range peers {
|
|
var exists bool
|
|
|
|
err = checkStmt.QueryRowContext(ctx, update.PeerIP).Scan(&exists)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to check peer %s: %w", update.PeerIP, err)
|
|
}
|
|
|
|
if exists {
|
|
_, err = updateStmt.ExecContext(ctx, update.PeerASN,
|
|
update.Timestamp, update.MessageType, update.PeerIP)
|
|
} else {
|
|
_, err = insertStmt.ExecContext(ctx,
|
|
generateUUID().String(), update.PeerIP, update.PeerASN,
|
|
update.Timestamp, update.Timestamp, update.MessageType)
|
|
}
|
|
|
|
if err != nil {
|
|
return fmt.Errorf("failed to update peer %s: %w", update.PeerIP, err)
|
|
}
|
|
|
|
if !exists {
|
|
newPeers++
|
|
}
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to commit transaction: %w", err)
|
|
}
|
|
|
|
d.counts.addPeers(newPeers)
|
|
|
|
return nil
|
|
}
|
|
|
|
// UpdatePeer updates or creates a BGP peer record
|
|
func (d *Database) UpdatePeer(
|
|
peerIP string, peerASN int, messageType string, timestamp time.Time,
|
|
) error {
|
|
d.lock("UpdatePeer")
|
|
defer d.unlock()
|
|
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
var exists bool
|
|
|
|
err = tx.QueryRow(
|
|
"SELECT EXISTS(SELECT 1 FROM bgp_peers WHERE peer_ip = ?)", peerIP,
|
|
).Scan(&exists)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if exists {
|
|
_, err = tx.Exec(
|
|
"UPDATE bgp_peers SET peer_asn = ?, last_seen = ?, "+
|
|
"last_message_type = ? WHERE peer_ip = ?",
|
|
peerASN, timestamp, messageType, peerIP,
|
|
)
|
|
} else {
|
|
_, err = tx.Exec(
|
|
"INSERT INTO bgp_peers (id, peer_ip, peer_asn, first_seen, "+
|
|
"last_seen, last_message_type) VALUES (?, ?, ?, ?, ?, ?)",
|
|
generateUUID().String(), peerIP, peerASN, timestamp, timestamp,
|
|
messageType,
|
|
)
|
|
}
|
|
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
d.logger.Error("Failed to commit transaction for peer update",
|
|
"peer_ip", peerIP,
|
|
"peer_asn", peerASN,
|
|
"error", err,
|
|
)
|
|
|
|
return err
|
|
}
|
|
|
|
if !exists {
|
|
d.counts.addPeers(1)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetStats returns database statistics
|
|
func (d *Database) GetStats() (Stats, error) {
|
|
return d.GetStatsContext(context.Background())
|
|
}
|
|
|
|
// GetStatsContext returns database statistics with context support.
|
|
//
|
|
// The row counts (ASNs, prefixes, peerings, peers, live routes) and the prefix
|
|
// distribution come from the in-memory counters, seeded at startup and kept
|
|
// current on every write. The oldest/newest route timestamps are read from the
|
|
// ends of the last_updated index, and the file size from a stat(). No part of
|
|
// the read scans a table or a whole index.
|
|
func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) {
|
|
var stats Stats
|
|
|
|
// Row counts and prefix distribution from memory, as a single consistent
|
|
// snapshot.
|
|
d.counts.fill(&stats)
|
|
|
|
// Database file size is a cheap stat() on the file.
|
|
fileInfo, err := os.Stat(d.path)
|
|
if err != nil {
|
|
d.logger.Warn("Failed to get database file size", "error", err)
|
|
} else {
|
|
stats.FileSizeBytes = fileInfo.Size()
|
|
}
|
|
|
|
// Oldest and newest route timestamps read one row from each end of the
|
|
// last_updated index (a log-time lookup, not a scan). Selecting the column
|
|
// directly lets the driver parse the DATETIME into time.Time; the old
|
|
// MIN/MAX union scan returned an untyped string that failed to scan and
|
|
// logged a warning on every call.
|
|
oldest, newest, err := d.routeTimestampRange(ctx)
|
|
if err != nil {
|
|
// Display-only fields; log but keep the rest of the stats.
|
|
d.logger.Warn("Failed to get route timestamps", "error", err)
|
|
} else {
|
|
stats.OldestRoute = oldest
|
|
stats.NewestRoute = newest
|
|
}
|
|
|
|
return stats, nil
|
|
}
|
|
|
|
// UpsertLiveRoute inserts or updates a live route
|
|
func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
|
|
d.lock("UpsertLiveRoute")
|
|
defer d.unlock()
|
|
|
|
pathJSON, err := json.Marshal(route.ASPath)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to encode AS path: %w", err)
|
|
}
|
|
|
|
originSQL, updateSQL := updateLiveRouteOriginV4SQL, updateLiveRouteV4SQL
|
|
insertSQL, lookupSQL := insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
|
|
|
|
if route.IPVersion == ipVersionV6 {
|
|
originSQL, updateSQL = updateLiveRouteOriginV6SQL, updateLiveRouteV6SQL
|
|
insertSQL, lookupSQL = insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
|
|
}
|
|
|
|
ctx := context.Background()
|
|
|
|
// The write lock is held, so no other writer can insert this key between the
|
|
// update and the insert even though they are separate autocommit statements.
|
|
setOrigin, err := d.writer.PrepareContext(ctx, originSQL)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to prepare origin statement: %w", err)
|
|
}
|
|
defer func() { _ = setOrigin.Close() }()
|
|
|
|
upd, err := d.writer.PrepareContext(ctx, updateSQL)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to prepare update statement: %w", err)
|
|
}
|
|
defer func() { _ = upd.Close() }()
|
|
|
|
ins, err := d.writer.PrepareContext(ctx, insertSQL)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to prepare insert statement: %w", err)
|
|
}
|
|
defer func() { _ = ins.Close() }()
|
|
|
|
has, err := d.writer.PrepareContext(ctx, lookupSQL)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to prepare prefix lookup statement: %w", err)
|
|
}
|
|
defer func() { _ = has.Close() }()
|
|
|
|
inserted, newPrefix, err := upsertRouteRow(ctx, setOrigin, upd, ins, has,
|
|
route, string(pathJSON))
|
|
if err != nil {
|
|
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
|
|
}
|
|
|
|
if inserted {
|
|
if route.IPVersion == ipVersionV4 {
|
|
d.counts.addRoutes(1, 0)
|
|
} else {
|
|
d.counts.addRoutes(0, 1)
|
|
}
|
|
}
|
|
|
|
if newPrefix {
|
|
if route.IPVersion == ipVersionV4 {
|
|
d.counts.addToDistribution([]int{route.MaskLength}, nil, 1)
|
|
} else {
|
|
d.counts.addToDistribution(nil, []int{route.MaskLength}, 1)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// DeleteLiveRoute deletes the live route the peer has for the prefix
|
|
func (d *Database) DeleteLiveRoute(prefix, peerIP string) error {
|
|
d.lock("DeleteLiveRoute")
|
|
defer d.unlock()
|
|
|
|
// Determine table based on prefix IP version
|
|
_, ipnet, err := net.ParseCIDR(prefix)
|
|
if err != nil {
|
|
return fmt.Errorf("invalid prefix format: %w", err)
|
|
}
|
|
|
|
isV4 := ipnet.IP.To4() != nil
|
|
|
|
// The delete and the prefix lookup after it share one transaction, so the
|
|
// in-memory counts change only once both have succeeded.
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to begin transaction: %w", err)
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
// Literal per-table queries (rather than one formatted with the table name)
|
|
// so the delete carries no dynamically built SQL.
|
|
deleteSQL := deleteLiveRouteV6SQL
|
|
if isV4 {
|
|
deleteSQL = deleteLiveRouteV4SQL
|
|
}
|
|
|
|
res, err := tx.Exec(deleteSQL, prefix, peerIP)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
affected, err := res.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if affected == 0 {
|
|
return nil
|
|
}
|
|
|
|
// The prefix leaves the distribution when no live route has it any more.
|
|
lookupSQL := prefixHasLiveRouteV6SQL
|
|
if isV4 {
|
|
lookupSQL = prefixHasLiveRouteV4SQL
|
|
}
|
|
|
|
var prefixHasRoute bool
|
|
|
|
err = tx.QueryRow(lookupSQL, prefix).Scan(&prefixHasRoute)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to commit transaction: %w", err)
|
|
}
|
|
|
|
if isV4 {
|
|
d.counts.addRoutes(-int(affected), 0)
|
|
} else {
|
|
d.counts.addRoutes(0, -int(affected))
|
|
}
|
|
|
|
if !prefixHasRoute {
|
|
maskLength, _ := ipnet.Mask.Size()
|
|
if isV4 {
|
|
d.counts.addToDistribution([]int{maskLength}, nil, -1)
|
|
} else {
|
|
d.counts.addToDistribution(nil, []int{maskLength}, -1)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetPrefixDistribution returns the distribution of unique prefixes by mask
|
|
// length
|
|
func (d *Database) GetPrefixDistribution() (
|
|
[]PrefixDistribution, []PrefixDistribution, error,
|
|
) {
|
|
return d.GetPrefixDistributionContext(context.Background())
|
|
}
|
|
|
|
// GetPrefixDistributionContext returns the distribution of unique prefixes by
|
|
// mask length with context support. It reads every live route, so the stats
|
|
// read does not call it; it seeds the in-memory distribution once at startup.
|
|
func (d *Database) GetPrefixDistributionContext(ctx context.Context) (
|
|
[]PrefixDistribution, []PrefixDistribution, error,
|
|
) {
|
|
var ipv4, ipv6 []PrefixDistribution
|
|
|
|
// IPv4 distribution - count unique prefixes from v4 table
|
|
query := `
|
|
SELECT mask_length, COUNT(DISTINCT prefix) as count
|
|
FROM live_routes_v4
|
|
GROUP BY mask_length
|
|
ORDER BY mask_length
|
|
`
|
|
|
|
rows4, err := d.db.QueryContext(ctx, query)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to query IPv4 distribution: %w", err)
|
|
}
|
|
defer func() {
|
|
if rows4 != nil {
|
|
_ = rows4.Close()
|
|
}
|
|
}()
|
|
|
|
for rows4.Next() {
|
|
var dist PrefixDistribution
|
|
|
|
err := rows4.Scan(&dist.MaskLength, &dist.Count)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to scan IPv4 distribution: %w",
|
|
err)
|
|
}
|
|
|
|
ipv4 = append(ipv4, dist)
|
|
}
|
|
|
|
// A read that stops partway ends the loop without an error of its own.
|
|
err = rows4.Err()
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to read IPv4 distribution: %w", err)
|
|
}
|
|
|
|
// IPv6 distribution - count unique prefixes from v6 table
|
|
query = `
|
|
SELECT mask_length, COUNT(DISTINCT prefix) as count
|
|
FROM live_routes_v6
|
|
GROUP BY mask_length
|
|
ORDER BY mask_length
|
|
`
|
|
|
|
rows6, err := d.db.QueryContext(ctx, query)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to query IPv6 distribution: %w", err)
|
|
}
|
|
defer func() {
|
|
if rows6 != nil {
|
|
_ = rows6.Close()
|
|
}
|
|
}()
|
|
|
|
for rows6.Next() {
|
|
var dist PrefixDistribution
|
|
|
|
err := rows6.Scan(&dist.MaskLength, &dist.Count)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to scan IPv6 distribution: %w",
|
|
err)
|
|
}
|
|
|
|
ipv6 = append(ipv6, dist)
|
|
}
|
|
|
|
err = rows6.Err()
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to read IPv6 distribution: %w", err)
|
|
}
|
|
|
|
return ipv4, ipv6, nil
|
|
}
|
|
|
|
// GetLiveRouteCounts returns the count of IPv4 and IPv6 routes
|
|
func (d *Database) GetLiveRouteCounts() (int, int, error) {
|
|
return d.GetLiveRouteCountsContext(context.Background())
|
|
}
|
|
|
|
// GetLiveRouteCountsContext returns the count of IPv4 and IPv6 routes with
|
|
// context support
|
|
func (d *Database) GetLiveRouteCountsContext(
|
|
ctx context.Context,
|
|
) (int, int, error) {
|
|
var ipv4Count, ipv6Count int
|
|
|
|
// Get IPv4 count from dedicated table
|
|
err := d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v4").
|
|
Scan(&ipv4Count)
|
|
if err != nil {
|
|
return 0, 0, fmt.Errorf("failed to count IPv4 routes: %w", err)
|
|
}
|
|
|
|
// Get IPv6 count from dedicated table
|
|
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v6").
|
|
Scan(&ipv6Count)
|
|
if err != nil {
|
|
return 0, 0, fmt.Errorf("failed to count IPv6 routes: %w", err)
|
|
}
|
|
|
|
return ipv4Count, ipv6Count, nil
|
|
}
|
|
|
|
// GetASDetails returns detailed information about an AS including prefixes
|
|
func (d *Database) GetASDetails(asn int) (*ASN, []LiveRoute, error) {
|
|
return d.GetASDetailsContext(context.Background(), asn)
|
|
}
|
|
|
|
// GetASDetailsContext returns detailed information about an AS including
|
|
// prefixes with context support
|
|
func (d *Database) GetASDetailsContext(
|
|
ctx context.Context, asn int,
|
|
) (*ASN, []LiveRoute, error) {
|
|
// Get AS information
|
|
var (
|
|
asnInfo ASN
|
|
handle, description sql.NullString
|
|
)
|
|
|
|
err := d.db.QueryRowContext(ctx,
|
|
"SELECT asn, handle, description, first_seen, last_seen "+
|
|
"FROM asns WHERE asn = ?",
|
|
asn,
|
|
).Scan(&asnInfo.ASN, &handle, &description, &asnInfo.FirstSeen,
|
|
&asnInfo.LastSeen)
|
|
if err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil, nil, fmt.Errorf("%w: AS%d", ErrNoRoute, asn)
|
|
}
|
|
|
|
return nil, nil, fmt.Errorf("failed to query AS: %w", err)
|
|
}
|
|
|
|
asnInfo.Handle = handle.String
|
|
asnInfo.Description = description.String
|
|
|
|
// Get prefixes announced by this AS from both tables
|
|
queryV4 := `
|
|
SELECT prefix, mask_length, MAX(last_updated) as last_updated
|
|
FROM live_routes_v4
|
|
WHERE origin_asn = ?
|
|
GROUP BY prefix, mask_length
|
|
`
|
|
|
|
allPrefixes, err := d.originPrefixes(ctx, queryV4, asn, ipVersionV4)
|
|
if err != nil {
|
|
return &asnInfo, nil, err
|
|
}
|
|
|
|
queryV6 := `
|
|
SELECT prefix, mask_length, MAX(last_updated) as last_updated
|
|
FROM live_routes_v6
|
|
WHERE origin_asn = ?
|
|
GROUP BY prefix, mask_length
|
|
`
|
|
|
|
prefixesV6, err := d.originPrefixes(ctx, queryV6, asn, ipVersionV6)
|
|
if err != nil {
|
|
return &asnInfo, allPrefixes, err
|
|
}
|
|
|
|
return &asnInfo, append(allPrefixes, prefixesV6...), nil
|
|
}
|
|
|
|
// ASPeer represents a peering relationship with another AS including handle,
|
|
// description, and timestamps.
|
|
//
|
|
//nolint:tagliatelle // field names are the existing JSON API
|
|
type ASPeer struct {
|
|
ASN int `json:"asn"`
|
|
Handle string `json:"handle"`
|
|
Description string `json:"description"`
|
|
FirstSeen time.Time `json:"first_seen"`
|
|
LastSeen time.Time `json:"last_seen"`
|
|
}
|
|
|
|
// GetASPeers returns all ASes that peer with the given AS
|
|
func (d *Database) GetASPeers(asn int) ([]ASPeer, error) {
|
|
return d.GetASPeersContext(context.Background(), asn)
|
|
}
|
|
|
|
// GetASPeersContext returns all ASes that peer with the given AS with context
|
|
// support
|
|
func (d *Database) GetASPeersContext(
|
|
ctx context.Context, asn int,
|
|
) ([]ASPeer, error) {
|
|
query := `
|
|
SELECT
|
|
CASE
|
|
WHEN p.as_a = ? THEN p.as_b
|
|
ELSE p.as_a
|
|
END as peer_asn,
|
|
COALESCE(a.handle, '') as handle,
|
|
COALESCE(a.description, '') as description,
|
|
p.first_seen,
|
|
p.last_seen
|
|
FROM peerings p
|
|
LEFT JOIN asns a ON a.asn = CASE
|
|
WHEN p.as_a = ? THEN p.as_b
|
|
ELSE p.as_a
|
|
END
|
|
WHERE p.as_a = ? OR p.as_b = ?
|
|
ORDER BY peer_asn
|
|
`
|
|
|
|
rows, err := d.db.QueryContext(ctx, query, asn, asn, asn, asn)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to query AS peers: %w", err)
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
var peers []ASPeer
|
|
|
|
for rows.Next() {
|
|
var peer ASPeer
|
|
|
|
err := rows.Scan(&peer.ASN, &peer.Handle, &peer.Description,
|
|
&peer.FirstSeen, &peer.LastSeen)
|
|
if err != nil {
|
|
d.logger.Error("Failed to scan peer row", "error", err, "asn", asn)
|
|
|
|
continue
|
|
}
|
|
|
|
peers = append(peers, peer)
|
|
}
|
|
|
|
err = rows.Err()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error iterating AS peers: %w", err)
|
|
}
|
|
|
|
return peers, nil
|
|
}
|
|
|
|
// GetPrefixDetails returns detailed information about a prefix
|
|
func (d *Database) GetPrefixDetails(
|
|
prefix string,
|
|
) ([]LiveRoute, map[int]ASN, error) {
|
|
return d.GetPrefixDetailsContext(context.Background(), prefix)
|
|
}
|
|
|
|
// GetPrefixDetailsContext returns the live routes for a prefix and, keyed by
|
|
// AS number, the handle and description of each AS that originates it.
|
|
func (d *Database) GetPrefixDetailsContext(
|
|
ctx context.Context, prefix string,
|
|
) ([]LiveRoute, map[int]ASN, error) {
|
|
// Determine if it's IPv4 or IPv6 by parsing the prefix
|
|
_, ipnet, err := net.ParseCIDR(prefix)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("invalid prefix format: %w", err)
|
|
}
|
|
|
|
tableName := tableLiveRoutesV4
|
|
ipVersion := ipVersionV4
|
|
|
|
if ipnet.IP.To4() == nil {
|
|
tableName = tableLiveRoutesV6
|
|
ipVersion = ipVersionV6
|
|
}
|
|
|
|
query := fmt.Sprintf(`
|
|
SELECT lr.origin_asn, lr.peer_ip, lr.as_path, lr.next_hop,
|
|
lr.last_updated, a.handle, a.description
|
|
FROM %s lr
|
|
LEFT JOIN asns a ON a.asn = lr.origin_asn
|
|
WHERE lr.prefix = ?
|
|
ORDER BY lr.origin_asn, lr.peer_ip
|
|
`, tableName)
|
|
|
|
rows, err := d.db.QueryContext(ctx, query, prefix)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to query prefix details: %w", err)
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
var routes []LiveRoute
|
|
|
|
origins := make(map[int]ASN)
|
|
|
|
for rows.Next() {
|
|
var (
|
|
route LiveRoute
|
|
pathJSON string
|
|
handle, description sql.NullString
|
|
)
|
|
|
|
err := rows.Scan(
|
|
&route.OriginASN, &route.PeerIP, &pathJSON, &route.NextHop,
|
|
&route.LastUpdated, &handle, &description,
|
|
)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
|
|
// Decode AS path
|
|
err = json.Unmarshal([]byte(pathJSON), &route.ASPath)
|
|
if err != nil {
|
|
route.ASPath = []int{}
|
|
}
|
|
|
|
route.Prefix = prefix
|
|
route.IPVersion = ipVersion
|
|
route.MaskLength, _ = ipnet.Mask.Size()
|
|
routes = append(routes, route)
|
|
origins[route.OriginASN] = ASN{
|
|
ASN: route.OriginASN,
|
|
Handle: handle.String,
|
|
Description: description.String,
|
|
}
|
|
}
|
|
|
|
err = rows.Err()
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to read prefix details: %w", err)
|
|
}
|
|
|
|
if len(routes) == 0 {
|
|
return nil, nil, fmt.Errorf("%w: %s", ErrNoRoute, prefix)
|
|
}
|
|
|
|
return routes, origins, nil
|
|
}
|
|
|
|
// GetRandomPrefixesByLength returns a random sample of prefixes with the
|
|
// specified mask length
|
|
func (d *Database) GetRandomPrefixesByLength(
|
|
maskLength, ipVersion, limit int,
|
|
) ([]LiveRoute, error) {
|
|
return d.GetRandomPrefixesByLengthContext(context.Background(),
|
|
maskLength, ipVersion, limit)
|
|
}
|
|
|
|
// GetRandomPrefixesByLengthContext returns limit prefixes with the given mask
|
|
// length picked at random, or all of them when there are fewer, each once with
|
|
// its most recently updated live route, for a limit up to
|
|
// maxPrefixesToShuffle. However many routes there are, it makes at most
|
|
// maxPrefixesToShuffle+1 index searches to list prefixes, at most
|
|
// sampleLookupsPerPrefix random lookups for each prefix asked for, and one
|
|
// index search for each listed prefix's newest route.
|
|
func (d *Database) GetRandomPrefixesByLengthContext(
|
|
ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error) {
|
|
tableName := tableLiveRoutesV4
|
|
if ipVersion == ipVersionV6 {
|
|
tableName = tableLiveRoutesV6
|
|
}
|
|
|
|
// Reading one prefix more than maxPrefixesToShuffle tells whether these
|
|
// are all the prefixes with this mask length.
|
|
prefixes, err := d.firstPrefixesByLength(ctx, tableName, maskLength,
|
|
maxPrefixesToShuffle+1)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
//nolint:gosec // The sample needs variety, not unpredictability.
|
|
rand.Shuffle(len(prefixes), func(i, j int) {
|
|
prefixes[i], prefixes[j] = prefixes[j], prefixes[i]
|
|
})
|
|
|
|
if len(prefixes) > maxPrefixesToShuffle {
|
|
prefixes, err = d.samplePrefixesByLength(ctx, tableName, maskLength,
|
|
limit, prefixes)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
prefixes = prefixes[:min(limit, len(prefixes))]
|
|
|
|
return d.newestRouteOfEachPrefix(ctx, tableName, ipVersion, prefixes)
|
|
}
|
|
|
|
// GetNextStaleASN returns a random ASN that needs WHOIS data refresh.
|
|
func (d *Database) GetNextStaleASN(
|
|
ctx context.Context, staleThreshold time.Duration,
|
|
) (int, error) {
|
|
cutoff := time.Now().Add(-staleThreshold)
|
|
|
|
// Select a random stale ASN using ORDER BY RANDOM()
|
|
query := `
|
|
SELECT asn FROM asns
|
|
WHERE whois_updated_at IS NULL
|
|
OR whois_updated_at < ?
|
|
ORDER BY RANDOM()
|
|
LIMIT 1
|
|
`
|
|
|
|
var asn int
|
|
|
|
err := d.db.QueryRowContext(ctx, query, cutoff).Scan(&asn)
|
|
if err != nil {
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return 0, ErrNoStaleASN
|
|
}
|
|
|
|
return 0, fmt.Errorf("failed to get stale ASN: %w", err)
|
|
}
|
|
|
|
return asn, nil
|
|
}
|
|
|
|
// WHOISStats contains statistics about WHOIS data freshness.
|
|
//
|
|
//nolint:tagliatelle // field names are the existing JSON API
|
|
type WHOISStats struct {
|
|
TotalASNs int `json:"total_asns"`
|
|
StaleASNs int `json:"stale_asns"`
|
|
FreshASNs int `json:"fresh_asns"`
|
|
NeverFetched int `json:"never_fetched"`
|
|
}
|
|
|
|
// GetWHOISStats returns statistics about WHOIS data freshness.
|
|
func (d *Database) GetWHOISStats(
|
|
ctx context.Context, staleThreshold time.Duration,
|
|
) (*WHOISStats, error) {
|
|
cutoff := time.Now().Add(-staleThreshold)
|
|
|
|
query := `
|
|
SELECT
|
|
COUNT(*) as total,
|
|
COALESCE(SUM(CASE WHEN whois_updated_at IS NULL
|
|
THEN 1 ELSE 0 END), 0) as never_fetched,
|
|
COALESCE(SUM(CASE WHEN whois_updated_at IS NOT NULL
|
|
AND whois_updated_at < ? THEN 1 ELSE 0 END), 0) as stale,
|
|
COALESCE(SUM(CASE WHEN whois_updated_at IS NOT NULL
|
|
AND whois_updated_at >= ? THEN 1 ELSE 0 END), 0) as fresh
|
|
FROM asns
|
|
`
|
|
|
|
var stats WHOISStats
|
|
|
|
err := d.db.QueryRowContext(ctx, query, cutoff, cutoff).Scan(
|
|
&stats.TotalASNs,
|
|
&stats.NeverFetched,
|
|
&stats.StaleASNs,
|
|
&stats.FreshASNs,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get WHOIS stats: %w", err)
|
|
}
|
|
|
|
return &stats, nil
|
|
}
|
|
|
|
// UpdateASNWHOIS updates an ASN record with WHOIS data.
|
|
func (d *Database) UpdateASNWHOIS(
|
|
ctx context.Context, update *ASNWHOISUpdate,
|
|
) error {
|
|
d.lock("UpdateASNWHOIS")
|
|
defer d.unlock()
|
|
|
|
query := `
|
|
UPDATE asns SET
|
|
as_name = ?,
|
|
org_name = ?,
|
|
org_id = ?,
|
|
address = ?,
|
|
country_code = ?,
|
|
abuse_email = ?,
|
|
abuse_phone = ?,
|
|
tech_email = ?,
|
|
tech_phone = ?,
|
|
rir = ?,
|
|
rir_registration_date = ?,
|
|
rir_last_modified = ?,
|
|
whois_raw = ?,
|
|
whois_updated_at = ?
|
|
WHERE asn = ?
|
|
`
|
|
|
|
_, err := d.writer.ExecContext(ctx, query,
|
|
update.ASName,
|
|
update.OrgName,
|
|
update.OrgID,
|
|
update.Address,
|
|
update.CountryCode,
|
|
update.AbuseEmail,
|
|
update.AbusePhone,
|
|
update.TechEmail,
|
|
update.TechPhone,
|
|
update.RIR,
|
|
update.RIRRegDate,
|
|
update.RIRLastMod,
|
|
update.WHOISRaw,
|
|
time.Now(),
|
|
update.ASN,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to update ASN WHOIS: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// GetIPInfo returns comprehensive IP information for the /ip endpoint.
|
|
func (d *Database) GetIPInfo(ip string) (*IPInfo, error) {
|
|
return d.GetIPInfoContext(context.Background(), ip)
|
|
}
|
|
|
|
// ipInfoQuery reads the live route for one prefix with its peer count, the
|
|
// prefix's first_seen and the origin AS's details. %[1]s is the live route
|
|
// table and %[2]s the prefix table of the address's IP version.
|
|
const ipInfoQuery = `
|
|
SELECT
|
|
lr.prefix,
|
|
lr.mask_length,
|
|
lr.origin_asn,
|
|
lr.last_updated,
|
|
(SELECT COUNT(DISTINCT peer_ip) FROM %[1]s
|
|
WHERE prefix = lr.prefix) as num_peers,
|
|
p.first_seen,
|
|
a.handle,
|
|
a.description,
|
|
a.as_name,
|
|
a.org_name,
|
|
a.org_id,
|
|
a.address,
|
|
a.country_code,
|
|
a.abuse_email,
|
|
a.rir,
|
|
a.whois_updated_at
|
|
FROM %[1]s lr
|
|
LEFT JOIN %[2]s p ON p.prefix = lr.prefix
|
|
LEFT JOIN asns a ON a.asn = lr.origin_asn
|
|
WHERE lr.prefix = ?
|
|
LIMIT 1
|
|
`
|
|
|
|
// GetIPInfoContext returns comprehensive IP information with context support.
|
|
//
|
|
// It finds the most specific live route for the address with point lookups on
|
|
// the prefix index: for each mask length from the longest (32 or 128) down to
|
|
// minLookupMaskLengthV4 or minLookupMaskLengthV6 it looks up the address's
|
|
// network prefix at that length, and the first length with a live route wins.
|
|
// That is at most 25 or 113 indexed lookups, however many routes are live. The
|
|
// prefix is looked up in the text form net/netip prints, which is the form the
|
|
// prefix handler stores.
|
|
func (d *Database) GetIPInfoContext(
|
|
ctx context.Context, ip string,
|
|
) (*IPInfo, error) {
|
|
addr, err := netip.ParseAddr(ip)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%w: %s", ErrInvalidIP, ip)
|
|
}
|
|
// An IPv4 address written in IPv6 form (::ffff:192.0.2.1) is an IPv4 lookup.
|
|
addr = addr.Unmap()
|
|
|
|
info := &IPInfo{IP: ip, IPVersion: ipVersionV6}
|
|
minMaskLength := minLookupMaskLengthV6
|
|
|
|
routesTable, prefixesTable := tableLiveRoutesV6, "prefixes_v6"
|
|
if addr.Is4() {
|
|
routesTable, prefixesTable = tableLiveRoutesV4, "prefixes_v4"
|
|
minMaskLength = minLookupMaskLengthV4
|
|
info.IPVersion = ipVersionV4
|
|
}
|
|
|
|
// Route info with peer count and prefix first_seen
|
|
query := fmt.Sprintf(ipInfoQuery, routesTable, prefixesTable)
|
|
|
|
stmt, err := d.db.PrepareContext(ctx, query)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to prepare route query: %w", err)
|
|
}
|
|
defer func() { _ = stmt.Close() }()
|
|
|
|
var (
|
|
handle, description, asName, orgName, orgID sql.NullString
|
|
address, countryCode, abuseEmail, rir sql.NullString
|
|
prefixFirstSeen, whoisUpdatedAt sql.NullTime
|
|
)
|
|
|
|
found := false
|
|
|
|
for bits := addr.BitLen(); bits >= minMaskLength; bits-- {
|
|
prefix := netip.PrefixFrom(addr, bits).Masked().String()
|
|
|
|
err := stmt.QueryRowContext(ctx, prefix).Scan(
|
|
&info.Netblock, &info.MaskLength, &info.ASN, &info.LastSeen,
|
|
&info.NumPeers, &prefixFirstSeen, &handle, &description,
|
|
&asName, &orgName, &orgID, &address, &countryCode, &abuseEmail,
|
|
&rir, &whoisUpdatedAt,
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
continue
|
|
}
|
|
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to query routes: %w", err)
|
|
}
|
|
|
|
found = true
|
|
|
|
break
|
|
}
|
|
|
|
if !found {
|
|
return nil, fmt.Errorf("%w for IP %s", ErrNoRoute, ip)
|
|
}
|
|
|
|
info.Handle = handle.String
|
|
info.Description = description.String
|
|
info.ASName = asName.String
|
|
info.OrgName = orgName.String
|
|
info.OrgID = orgID.String
|
|
info.Address = address.String
|
|
info.CountryCode = countryCode.String
|
|
info.AbuseEmail = abuseEmail.String
|
|
info.RIR = rir.String
|
|
|
|
if prefixFirstSeen.Valid {
|
|
info.FirstSeen = prefixFirstSeen.Time
|
|
}
|
|
|
|
// Check if WHOIS data needs refresh (never fetched or older than 30 days).
|
|
// A route with origin 0 has no known origin, so no AS to look up.
|
|
const staleThreshold = 30 * 24 * time.Hour
|
|
|
|
info.NeedsWHOISRefresh = info.ASN != 0 && (!whoisUpdatedAt.Valid ||
|
|
time.Since(whoisUpdatedAt.Time) > staleThreshold)
|
|
|
|
return info, nil
|
|
}
|
|
|
|
// Vacuum runs incremental vacuum to return up to 1000 unused pages to the
|
|
// filesystem. Like Analyze and Checkpoint, it holds the database lock, so it
|
|
// and the batch writes wait for each other instead of failing on SQLite's
|
|
// write lock, and returns ctx's error without running if ctx ended while it
|
|
// waited: the writing connection it uses does not check ctx itself.
|
|
func (d *Database) Vacuum(ctx context.Context) error {
|
|
// Free up to 1000 pages per call (~4MB with default 4KB page size)
|
|
// This keeps each vacuum operation quick
|
|
const pagesToFree = 1000
|
|
|
|
d.lock("Vacuum")
|
|
defer d.unlock()
|
|
|
|
err := ctx.Err()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to run incremental vacuum: %w", err)
|
|
}
|
|
|
|
rows, err := d.writer.QueryContext(ctx,
|
|
fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree))
|
|
if err != nil {
|
|
return fmt.Errorf("failed to run incremental vacuum: %w", err)
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
// SQLite frees one page each time the statement steps, and each step
|
|
// returns a row, so every row must be read for the pragma to free more
|
|
// than one page.
|
|
for rows.Next() {
|
|
}
|
|
|
|
err = rows.Err()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to run incremental vacuum: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Analyze updates the query planner statistics. It runs ANALYZE, through
|
|
// PRAGMA optimize, only on the tables that have an index without statistics or
|
|
// whose size has changed tenfold since they were last analyzed: even with
|
|
// analysisLimit, ANALYZE counts every entry of each index, so it takes longer
|
|
// as the tables grow. It holds the database lock.
|
|
func (d *Database) Analyze(ctx context.Context) error {
|
|
d.lock("Analyze")
|
|
defer d.unlock()
|
|
|
|
err := ctx.Err()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to analyze database: %w", err)
|
|
}
|
|
|
|
// analysis_limit is a setting of the connection, so it is set in the same
|
|
// call as PRAGMA optimize, which runs both statements on one connection.
|
|
_, err = d.writer.ExecContext(ctx, fmt.Sprintf(
|
|
"PRAGMA analysis_limit=%d; PRAGMA optimize=%#x",
|
|
analysisLimit, optimizeAllTables))
|
|
if err != nil {
|
|
return fmt.Errorf("failed to analyze database: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Checkpoint runs a WAL checkpoint to transfer data from the WAL to the main database.
|
|
// Uses TRUNCATE mode which blocks writers briefly but ensures complete checkpoint,
|
|
// keeping the WAL small for fast read performance. It holds the database lock.
|
|
func (d *Database) Checkpoint(ctx context.Context) error {
|
|
d.lock("Checkpoint")
|
|
defer d.unlock()
|
|
|
|
err := ctx.Err()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to checkpoint WAL: %w", err)
|
|
}
|
|
|
|
_, err = d.writer.ExecContext(ctx, "PRAGMA wal_checkpoint(TRUNCATE)")
|
|
if err != nil {
|
|
return fmt.Errorf("failed to checkpoint WAL: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Ping checks if the database connection is alive with a lightweight query.
|
|
func (d *Database) Ping(ctx context.Context) error {
|
|
var result int
|
|
|
|
err := d.db.QueryRowContext(ctx, "SELECT 1").Scan(&result)
|
|
if err != nil {
|
|
return fmt.Errorf("database ping failed: %w", err)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// The debug messages lock logs, with the operation's name, when it starts to
|
|
// wait for the database mutex and when it holds it.
|
|
const (
|
|
lockWaitMessage = "Acquiring database lock"
|
|
lockHeldMessage = "Database lock acquired"
|
|
)
|
|
|
|
// lock acquires the database mutex and logs debug information
|
|
func (d *Database) lock(operation string) {
|
|
// Get caller information
|
|
_, file, line, _ := runtime.Caller(1)
|
|
caller := fmt.Sprintf("%s:%d", filepath.Base(file), line)
|
|
|
|
d.logger.Debug(lockWaitMessage, "operation", operation, "caller", caller)
|
|
|
|
d.mu.Lock()
|
|
d.lockedAt = time.Now()
|
|
d.lockedBy = fmt.Sprintf("%s (%s)", operation, caller)
|
|
|
|
d.logger.Debug(lockHeldMessage, "operation", operation, "caller", caller)
|
|
}
|
|
|
|
// unlock releases the database mutex and logs debug information including
|
|
// hold duration
|
|
func (d *Database) unlock() {
|
|
holdDuration := time.Since(d.lockedAt)
|
|
lockedBy := d.lockedBy
|
|
|
|
d.lockedAt = time.Time{}
|
|
d.lockedBy = ""
|
|
d.mu.Unlock()
|
|
|
|
d.logger.Debug("Database lock released",
|
|
"held_by", lockedBy, "duration_ms", holdDuration.Milliseconds())
|
|
}
|
|
|
|
// beginTx starts a new transaction on the writing connection, with logging
|
|
func (d *Database) beginTx() (*loggingTx, error) {
|
|
tx, err := d.writer.BeginTx(context.Background(), nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &loggingTx{Tx: tx, logger: d.logger}, nil
|
|
}
|
|
|
|
// rollback rolls back tx unless it has already been committed, and logs a
|
|
// failure. Every write defers it right after beginning its transaction.
|
|
func (d *Database) rollback(tx *loggingTx) {
|
|
err := tx.Rollback()
|
|
if err != nil && !errors.Is(err, sql.ErrTxDone) {
|
|
d.logger.Error("Failed to rollback transaction", "error", err)
|
|
}
|
|
}
|
|
|
|
// routeTimestampRange returns the earliest and latest last_updated across both
|
|
// live route tables, or nil values when both tables are empty. Each query reads
|
|
// one row from an end of the last_updated index rather than scanning the tables.
|
|
func (d *Database) routeTimestampRange(
|
|
ctx context.Context,
|
|
) (*time.Time, *time.Time, error) {
|
|
var oldest, newest *time.Time
|
|
|
|
oldestV4, ok, err := d.scanRouteTimestamp(ctx, "SELECT last_updated "+
|
|
"FROM live_routes_v4 ORDER BY last_updated ASC LIMIT 1")
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
if ok {
|
|
oldest = &oldestV4
|
|
}
|
|
|
|
oldestV6, ok, err := d.scanRouteTimestamp(ctx, "SELECT last_updated "+
|
|
"FROM live_routes_v6 ORDER BY last_updated ASC LIMIT 1")
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
if ok && (oldest == nil || oldestV6.Before(*oldest)) {
|
|
oldest = &oldestV6
|
|
}
|
|
|
|
newestV4, ok, err := d.scanRouteTimestamp(ctx, "SELECT last_updated "+
|
|
"FROM live_routes_v4 ORDER BY last_updated DESC LIMIT 1")
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
if ok {
|
|
newest = &newestV4
|
|
}
|
|
|
|
newestV6, ok, err := d.scanRouteTimestamp(ctx, "SELECT last_updated "+
|
|
"FROM live_routes_v6 ORDER BY last_updated DESC LIMIT 1")
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
if ok && (newest == nil || newestV6.After(*newest)) {
|
|
newest = &newestV6
|
|
}
|
|
|
|
return oldest, newest, nil
|
|
}
|
|
|
|
// scanRouteTimestamp runs a single-row timestamp query. The bool is false when
|
|
// the table is empty. The query selects the last_updated column directly so the
|
|
// driver parses the DATETIME value into a time.Time.
|
|
func (d *Database) scanRouteTimestamp(
|
|
ctx context.Context, query string,
|
|
) (time.Time, bool, error) {
|
|
var ts time.Time
|
|
|
|
err := d.db.QueryRowContext(ctx, query).Scan(&ts)
|
|
switch {
|
|
case errors.Is(err, sql.ErrNoRows):
|
|
return time.Time{}, false, nil
|
|
case err != nil:
|
|
return time.Time{}, false, err
|
|
default:
|
|
return ts, true, nil
|
|
}
|
|
}
|
|
|
|
// originPrefixes runs query, which selects the prefix, mask length and latest
|
|
// last_updated of every prefix asn originates in the live route table of
|
|
// ipVersion. Rows that fail to scan or whose timestamp does not parse are
|
|
// logged and skipped.
|
|
func (d *Database) originPrefixes(
|
|
ctx context.Context, query string, asn, ipVersion int,
|
|
) ([]LiveRoute, error) {
|
|
family := ipv4Family
|
|
if ipVersion == ipVersionV6 {
|
|
family = ipv6Family
|
|
}
|
|
|
|
rows, err := d.db.QueryContext(ctx, query, asn)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to query %s prefixes: %w", family, err)
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
var routes []LiveRoute
|
|
|
|
for rows.Next() {
|
|
var (
|
|
route LiveRoute
|
|
lastUpdatedStr string
|
|
)
|
|
|
|
err := rows.Scan(&route.Prefix, &route.MaskLength, &lastUpdatedStr)
|
|
if err != nil {
|
|
d.logger.Error("Failed to scan "+family+" prefix row",
|
|
"error", err, "asn", asn)
|
|
|
|
continue
|
|
}
|
|
|
|
// Parse the timestamp string
|
|
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05-07:00",
|
|
lastUpdatedStr)
|
|
if err != nil {
|
|
// Try without timezone
|
|
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05",
|
|
lastUpdatedStr)
|
|
if err != nil {
|
|
d.logger.Error("Failed to parse timestamp",
|
|
"error", err, "timestamp", lastUpdatedStr)
|
|
|
|
continue
|
|
}
|
|
}
|
|
|
|
route.OriginASN = asn
|
|
route.IPVersion = ipVersion
|
|
routes = append(routes, route)
|
|
}
|
|
|
|
err = rows.Err()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to read %s prefixes: %w", family, err)
|
|
}
|
|
|
|
return routes, nil
|
|
}
|
|
|
|
// firstPrefixesByLength returns the first n prefixes, in text order, with the
|
|
// given mask length. Each prefix is one search of the mask length and prefix
|
|
// index for the prefix after the previous one, so the read does not grow with
|
|
// the number of peers that carry a prefix, as a SELECT DISTINCT would.
|
|
func (d *Database) firstPrefixesByLength(
|
|
ctx context.Context, tableName string, maskLength, n int) ([]string, error) {
|
|
//nolint:gosec // Table name is hardcoded based on IP version
|
|
query := fmt.Sprintf(`
|
|
SELECT prefix FROM %s
|
|
WHERE mask_length = ? AND prefix > ?
|
|
ORDER BY prefix
|
|
LIMIT 1
|
|
`, tableName)
|
|
|
|
stmt, err := d.db.PrepareContext(ctx, query)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to prepare prefix query: %w", err)
|
|
}
|
|
defer func() { _ = stmt.Close() }()
|
|
|
|
var prefixes []string
|
|
|
|
previous := "" // every prefix sorts after the empty string
|
|
|
|
for len(prefixes) < n {
|
|
var prefix string
|
|
|
|
err := stmt.QueryRowContext(ctx, maskLength, previous).Scan(&prefix)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
break
|
|
}
|
|
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to list prefixes: %w", err)
|
|
}
|
|
|
|
prefixes = append(prefixes, prefix)
|
|
previous = prefix
|
|
}
|
|
|
|
return prefixes, nil
|
|
}
|
|
|
|
// samplePrefixesByLength picks limit distinct prefixes with the given mask
|
|
// length at random, in at most sampleLookupsPerPrefix lookups for each prefix
|
|
// asked for. ORDER BY RANDOM() would read every route with that mask length.
|
|
// Instead each lookup draws a row id between those of the first and last route
|
|
// with that mask length and takes the prefix of the first such route at or
|
|
// after it. SQLite keeps the row id at the end of every index entry, so the
|
|
// mask length index is ordered by mask length, then row id, and the lookup is
|
|
// one search of it.
|
|
//
|
|
// The sample is uneven, as the page needs a varied sample, not an even one: a
|
|
// route is picked in proportion to the row ids between it and the previous
|
|
// route with that mask length, including the row ids of routes with other mask
|
|
// lengths and of withdrawn routes. A route that follows a long run of
|
|
// withdrawn routes is picked far more often than one that directly follows
|
|
// another route with that mask length.
|
|
//
|
|
// Should the lookups keep landing on prefixes already picked, the rest are
|
|
// taken from shuffled, prefixes of that length in random order.
|
|
func (d *Database) samplePrefixesByLength(
|
|
ctx context.Context, tableName string, maskLength, limit int,
|
|
shuffled []string,
|
|
) ([]string, error) {
|
|
//nolint:gosec // Table name is hardcoded based on IP version
|
|
rangeQuery := fmt.Sprintf(`
|
|
SELECT
|
|
(SELECT MIN(rowid) FROM %[1]s WHERE mask_length = ?),
|
|
(SELECT MAX(rowid) FROM %[1]s WHERE mask_length = ?)
|
|
`, tableName)
|
|
|
|
var first, last sql.NullInt64
|
|
|
|
err := d.db.QueryRowContext(ctx, rangeQuery, maskLength, maskLength).
|
|
Scan(&first, &last)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to read row id range: %w", err)
|
|
}
|
|
|
|
if !first.Valid {
|
|
// Every route with this mask length was withdrawn since they were listed.
|
|
return nil, nil
|
|
}
|
|
|
|
//nolint:gosec // Table name is hardcoded based on IP version
|
|
lookupQuery := fmt.Sprintf(`
|
|
SELECT prefix FROM %s
|
|
WHERE mask_length = ? AND rowid >= ?
|
|
ORDER BY rowid
|
|
LIMIT 1
|
|
`, tableName)
|
|
|
|
stmt, err := d.db.PrepareContext(ctx, lookupQuery)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to prepare prefix lookup: %w", err)
|
|
}
|
|
defer func() { _ = stmt.Close() }()
|
|
|
|
picked := make(map[string]bool, limit)
|
|
prefixes := make([]string, 0, limit)
|
|
pick := func(prefix string) {
|
|
if !picked[prefix] {
|
|
picked[prefix] = true
|
|
prefixes = append(prefixes, prefix)
|
|
}
|
|
}
|
|
|
|
maxLookups := limit * sampleLookupsPerPrefix
|
|
for lookups := 0; lookups < maxLookups && len(prefixes) < limit; lookups++ {
|
|
//nolint:gosec // The sample needs variety, not unpredictability.
|
|
rowID := first.Int64 + rand.Int64N(last.Int64-first.Int64+1)
|
|
|
|
var prefix string
|
|
|
|
err := stmt.QueryRowContext(ctx, maskLength, rowID).Scan(&prefix)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
// The routes from rowID on were withdrawn since the range was read.
|
|
continue
|
|
}
|
|
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to look up prefix: %w", err)
|
|
}
|
|
|
|
pick(prefix)
|
|
}
|
|
|
|
for _, prefix := range shuffled {
|
|
if len(prefixes) == limit {
|
|
break
|
|
}
|
|
|
|
pick(prefix)
|
|
}
|
|
|
|
return prefixes, nil
|
|
}
|
|
|
|
// newestRouteOfEachPrefix returns the most recently updated live route of each
|
|
// prefix, one search of the prefix and last_updated index per prefix however
|
|
// many peers carry it. A prefix whose routes were all withdrawn since it was
|
|
// picked is left out.
|
|
func (d *Database) newestRouteOfEachPrefix(
|
|
ctx context.Context, tableName string, ipVersion int, prefixes []string,
|
|
) ([]LiveRoute, error) {
|
|
//nolint:gosec // Table name is hardcoded based on IP version
|
|
query := fmt.Sprintf(`
|
|
SELECT prefix, mask_length, origin_asn, as_path, peer_ip, last_updated
|
|
FROM %s
|
|
WHERE prefix = ?
|
|
ORDER BY last_updated DESC
|
|
LIMIT 1
|
|
`, tableName)
|
|
|
|
stmt, err := d.db.PrepareContext(ctx, query)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to prepare route query: %w", err)
|
|
}
|
|
defer func() { _ = stmt.Close() }()
|
|
|
|
routes := make([]LiveRoute, 0, len(prefixes))
|
|
|
|
for _, prefix := range prefixes {
|
|
route := LiveRoute{IPVersion: ipVersion}
|
|
|
|
var pathJSON string
|
|
|
|
err := stmt.QueryRowContext(ctx, prefix).Scan(
|
|
&route.Prefix,
|
|
&route.MaskLength,
|
|
&route.OriginASN,
|
|
&pathJSON,
|
|
&route.PeerIP,
|
|
&route.LastUpdated,
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
continue
|
|
}
|
|
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to query route of %s: %w", prefix, err)
|
|
}
|
|
|
|
err = json.Unmarshal([]byte(pathJSON), &route.ASPath)
|
|
if err != nil {
|
|
route.ASPath = []int{}
|
|
}
|
|
|
|
routes = append(routes, route)
|
|
}
|
|
|
|
return routes, nil
|
|
}
|