clawbot
a173f4d159
check / check (push) Canceled after 0s
When the delete of a peer's live routes after its session went down failed, the down was dropped and the routes stayed. The prefix handler now keeps such a peer, once, with the updates it sent after the down, for the next flush, which retries the delete before it writes them; each failure is logged. At most the last 25,000 of those updates are kept, outside the batch, so they do not start a flush. The delete runs in chunks of at most 10,000 routes per table, taken in prefix order through the peer's key, each chunk in its own transaction, so the database lock is released between chunks and a failure undoes at most one chunk. Model: opus-5-5
2711 lines
74 KiB
Go
2711 lines
74 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
|
|
}
|
|
|
|
// peerRouteChunkSize is the most routes of a peer that one transaction of
|
|
// DeleteLiveRoutesOfPeer deletes from each route table. A peer can have a
|
|
// million routes; deleted in one transaction, they would hold the database lock
|
|
// for most of a minute.
|
|
const peerRouteChunkSize = 10000
|
|
|
|
// Deleting a chunk of a peer's live routes. The first query of each table
|
|
// selects the peer's first routes in prefix order, through the
|
|
// UNIQUE(peer_ip, prefix) key, each with its mask length and whether another
|
|
// peer has a live route for its prefix: the delete takes the prefixes no other
|
|
// peer has out of the prefix distribution. The second query deletes the peer's
|
|
// routes up to the last prefix the first one selected.
|
|
const (
|
|
peerRouteChunkV4SQL = `SELECT r.prefix, r.mask_length,
|
|
EXISTS (SELECT 1 FROM live_routes_v4 AS o
|
|
WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)
|
|
FROM live_routes_v4 AS r WHERE r.peer_ip = ?
|
|
ORDER BY r.prefix LIMIT ?`
|
|
deletePeerRouteChunkV4SQL = `DELETE FROM live_routes_v4
|
|
WHERE peer_ip = ? AND prefix <= ?`
|
|
|
|
peerRouteChunkV6SQL = `SELECT r.prefix, r.mask_length,
|
|
EXISTS (SELECT 1 FROM live_routes_v6 AS o
|
|
WHERE o.prefix = r.prefix AND o.peer_ip != r.peer_ip)
|
|
FROM live_routes_v6 AS r WHERE r.peer_ip = ?
|
|
ORDER BY r.prefix LIMIT ?`
|
|
deletePeerRouteChunkV6SQL = `DELETE FROM live_routes_v6
|
|
WHERE peer_ip = ? AND prefix <= ?`
|
|
)
|
|
|
|
// deletePeerRoutes deletes the first peerRouteChunkSize live routes of peerIP,
|
|
// or as many as it has, 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, chunkSQL, deleteSQL, peerIP string,
|
|
) (int64, []int, error) {
|
|
rows, err := tx.QueryContext(ctx, chunkSQL, peerIP, peerRouteChunkSize)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
defer func() { _ = rows.Close() }()
|
|
|
|
var (
|
|
selected int
|
|
lastPrefix string
|
|
goneMaskLengths []int
|
|
)
|
|
|
|
for rows.Next() {
|
|
var (
|
|
maskLength int
|
|
otherPeerHasARoute bool
|
|
)
|
|
|
|
err := rows.Scan(&lastPrefix, &maskLength, &otherPeerHasARoute)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
selected++
|
|
|
|
if !otherPeerHasARoute {
|
|
goneMaskLengths = append(goneMaskLengths, maskLength)
|
|
}
|
|
}
|
|
|
|
err = rows.Err()
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
if selected == 0 {
|
|
return 0, nil, nil
|
|
}
|
|
|
|
res, err := tx.ExecContext(ctx, deleteSQL, peerIP, lastPrefix)
|
|
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,
|
|
// as when the peer's BGP session goes down. It deletes them a chunk at a time
|
|
// with deletePeerRouteChunk, each chunk in a transaction of its own, and
|
|
// releases the database lock between chunks so other writes go on. It returns
|
|
// how many routes it deleted; after an error, the chunks it deleted before stay
|
|
// deleted.
|
|
func (d *Database) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
|
|
total := 0
|
|
|
|
for {
|
|
deletedV4, deletedV6, err := d.deletePeerRouteChunk(peerIP)
|
|
if err != nil {
|
|
return total, err
|
|
}
|
|
|
|
total += deletedV4 + deletedV6
|
|
|
|
if deletedV4 < peerRouteChunkSize && deletedV6 < peerRouteChunkSize {
|
|
return total, 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)
|
|
}
|
|
}
|
|
|
|
// deletePeerRouteChunk deletes, in one transaction, the first
|
|
// peerRouteChunkSize IPv4 and the first peerRouteChunkSize IPv6 live routes of
|
|
// the peer, or as many as it has. It returns how many of each it deleted.
|
|
func (d *Database) deletePeerRouteChunk(peerIP string) (int, int, error) {
|
|
d.lock("DeleteLiveRoutesOfPeer")
|
|
defer d.unlock()
|
|
|
|
tx, err := d.beginTx()
|
|
if err != nil {
|
|
return 0, 0, fmt.Errorf("failed to begin transaction: %w", err)
|
|
}
|
|
defer d.rollback(tx)
|
|
|
|
ctx := context.Background()
|
|
|
|
deletedV4, goneMaskLengthsV4, err := deletePeerRoutes(ctx, tx,
|
|
peerRouteChunkV4SQL, deletePeerRouteChunkV4SQL, peerIP)
|
|
if err != nil {
|
|
return 0, 0, fmt.Errorf("failed to delete the IPv4 routes of peer %s: %w",
|
|
peerIP, err)
|
|
}
|
|
|
|
deletedV6, goneMaskLengthsV6, err := deletePeerRoutes(ctx, tx,
|
|
peerRouteChunkV6SQL, deletePeerRouteChunkV6SQL, peerIP)
|
|
if err != nil {
|
|
return 0, 0, fmt.Errorf("failed to delete the IPv6 routes of peer %s: %w",
|
|
peerIP, err)
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return 0, 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), int(deletedV6), nil
|
|
}
|
|
|
|
// 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
|
|
}
|