Serve /api/v1/stats counts from realtime in-memory counters (closes #27) #29

Merged
clawbot merged 1 commits from issue-27-realtime-stats into next 2026-09-22 09:41:16 +02:00
6 changed files with 812 additions and 222 deletions
Showing only changes of commit 698390f66c - Show all commits
+4
View File
@@ -23,6 +23,10 @@ runs make check on main.
# Completed Steps
- 2026-09-22: realtime in-memory database statistics: counts seeded at
startup and adjusted on every write, oldest/newest route timestamps via
index-end lookups; `/api/v1/stats` no longer scans the tables (closes
#27)
- 2026-07-07 Adopted scripts-to-rule-them-all: `script/` entrypoints,
Makefile shims, README Entrypoints section
- 2026-02-22: repo policy compliance: required policy files, .gitignore
+146
View File
@@ -0,0 +1,146 @@
package database
import (
"context"
"fmt"
"sync"
)
// liveCounts holds the running row counts that the stats endpoints report. They
// are seeded once at startup from the tables and then adjusted on every write,
// so a stats read serves them from memory instead of running a COUNT(*) over
// each table. Those scans, once the database passed a few GiB, took the whole
// request timeout and made /api/v1/stats return 500 (issue 27).
//
// A single mutex guards all fields so the stats reader takes a consistent
// snapshot at one instant and writers, which already run under the database
// write lock, adjust the counts after their transaction commits.
type liveCounts struct {
mu sync.RWMutex
asns int
prefixesV4 int
prefixesV6 int
peerings int
peers int
routesV4 int
routesV6 int
}
// seed sets every count to the value read from the tables at startup. It runs
// before any writer, so it needs no coordination with the adjust methods.
func (c *liveCounts) seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6 int) {
c.mu.Lock()
defer c.mu.Unlock()
c.asns = asns
c.prefixesV4 = prefixesV4
c.prefixesV6 = prefixesV6
c.peerings = peerings
c.peers = peers
c.routesV4 = routesV4
c.routesV6 = routesV6
}
// addASNs adds n to the ASN count.
func (c *liveCounts) addASNs(n int) {
c.mu.Lock()
c.asns += n
c.mu.Unlock()
}
// addPrefixes adds to the IPv4 and IPv6 prefix counts.
func (c *liveCounts) addPrefixes(v4, v6 int) {
c.mu.Lock()
c.prefixesV4 += v4
c.prefixesV6 += v6
c.mu.Unlock()
}
// addPeerings adds n to the peering count.
func (c *liveCounts) addPeerings(n int) {
c.mu.Lock()
c.peerings += n
c.mu.Unlock()
}
// addPeers adds n to the BGP peer count.
func (c *liveCounts) addPeers(n int) {
c.mu.Lock()
c.peers += n
c.mu.Unlock()
}
// addRoutes adds to the IPv4 and IPv6 live-route counts. Deletions pass
// negative values.
func (c *liveCounts) addRoutes(v4, v6 int) {
c.mu.Lock()
c.routesV4 += v4
c.routesV6 += v6
c.mu.Unlock()
}
// fill copies the counts into a Stats, including the derived totals, under a
// single read lock so the reader sees one consistent snapshot.
func (c *liveCounts) fill(s *Stats) {
c.mu.RLock()
defer c.mu.RUnlock()
s.ASNs = c.asns
s.IPv4Prefixes = c.prefixesV4
s.IPv6Prefixes = c.prefixesV6
s.Prefixes = c.prefixesV4 + c.prefixesV6
s.Peerings = c.peerings
s.Peers = c.peers
s.IPv4Routes = c.routesV4
s.IPv6Routes = c.routesV6
s.LiveRoutes = c.routesV4 + c.routesV6
}
// countRows returns the number of rows in the named table. It is used only at
// startup to seed the in-memory counters, so a full COUNT(*) is acceptable.
func (d *Database) countRows(ctx context.Context, table string) (int, error) {
var n int
// table is one of a fixed set of literals below, never external input.
if err := d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+table).Scan(&n); err != nil {
return 0, fmt.Errorf("failed to count %s: %w", table, err)
}
return n, nil
}
// seedCounts reads the current row counts from the tables into the in-memory
// counters. It runs once at startup, before the streamer begins writing.
func (d *Database) seedCounts(ctx context.Context) error {
asns, err := d.countRows(ctx, "asns")
if err != nil {
return err
}
prefixesV4, err := d.countRows(ctx, "prefixes_v4")
if err != nil {
return err
}
prefixesV6, err := d.countRows(ctx, "prefixes_v6")
if err != nil {
return err
}
peerings, err := d.countRows(ctx, "peerings")
if err != nil {
return err
}
peers, err := d.countRows(ctx, "bgp_peers")
if err != nil {
return err
}
routesV4, err := d.countRows(ctx, "live_routes_v4")
if err != nil {
return err
}
routesV6, err := d.countRows(ctx, "live_routes_v6")
if err != nil {
return err
}
d.counts.seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6)
return nil
}
+335
View File
@@ -0,0 +1,335 @@
package database
import (
"context"
"sync"
"testing"
"time"
"git.eeqj.de/sneak/routewatch/internal/config"
"git.eeqj.de/sneak/routewatch/internal/logger"
"github.com/google/uuid"
)
// mkV4Route builds an IPv4 live route with its range columns populated.
func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute {
t.Helper()
start, end, err := CalculateIPv4Range(prefix)
if err != nil {
t.Fatalf("CalculateIPv4Range(%s): %v", prefix, err)
}
return &LiveRoute{
ID: uuid.New(),
Prefix: prefix,
MaskLength: 24,
IPVersion: ipVersionV4,
OriginASN: asn,
PeerIP: "192.0.2.1",
ASPath: []int{asn},
NextHop: "192.0.2.254",
LastUpdated: ts,
V4IPStart: &start,
V4IPEnd: &end,
}
}
// mkV6Route builds an IPv6 live route.
func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute {
return &LiveRoute{
ID: uuid.New(),
Prefix: prefix,
MaskLength: 32,
IPVersion: ipVersionV6,
OriginASN: asn,
PeerIP: "2001:db8::1",
ASPath: []int{asn},
NextHop: "2001:db8::ffff",
LastUpdated: ts,
}
}
// TestLiveCountsTrackWritesInRealtime checks that the stats counts start at
// zero, reflect each write the moment it commits (no recompute, no timer), do
// not move when a route is merely re-announced, and drop when a route is
// deleted. These counts are what /api/v1/stats reports; before this change the
// endpoint recomputed them with a COUNT(*) over each table on every request.
func TestLiveCountsTrackWritesInRealtime(t *testing.T) {
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ctx := context.Background()
empty, err := db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("GetStatsContext on empty database: %v", err)
}
if empty.ASNs != 0 || empty.Prefixes != 0 || empty.Peerings != 0 ||
empty.Peers != 0 || empty.LiveRoutes != 0 {
t.Fatalf("empty database counts nonzero: %+v", empty)
}
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
if err := db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts}); err != nil {
t.Fatalf("GetOrCreateASNBatch: %v", err)
}
if err := db.UpdatePrefixesBatch(map[string]time.Time{
"198.51.100.0/24": ts,
"2001:db8::/32": ts,
}); err != nil {
t.Fatalf("UpdatePrefixesBatch: %v", err)
}
if err := db.UpdatePeerBatch(map[string]PeerUpdate{
"192.0.2.1": {PeerIP: "192.0.2.1", PeerASN: 64500, MessageType: "UPDATE", Timestamp: ts},
}); err != nil {
t.Fatalf("UpdatePeerBatch: %v", err)
}
if err := db.RecordPeering(64500, 64501, ts); err != nil {
t.Fatalf("RecordPeering: %v", err)
}
routes := []*LiveRoute{
mkV4Route(t, "198.51.100.0/24", 64500, ts),
mkV4Route(t, "203.0.113.0/24", 64501, ts.Add(time.Minute)),
mkV6Route("2001:db8::/32", 64502, ts.Add(2*time.Minute)),
}
if err := db.UpsertLiveRouteBatch(routes); err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
stats, err := db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("GetStatsContext: %v", err)
}
assertCounts(t, "after inserts", stats, wantCounts{
asns: 2, prefixes: 2, peerings: 1, peers: 1,
ipv4Routes: 2, ipv6Routes: 1, liveRoutes: 3,
})
// Re-announcing the same routes is an update, not an insert: counts hold.
if err := db.UpsertLiveRouteBatch(routes); err != nil {
t.Fatalf("UpsertLiveRouteBatch (re-announce): %v", err)
}
stats, err = db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("GetStatsContext: %v", err)
}
assertCounts(t, "after re-announce", stats, wantCounts{
asns: 2, prefixes: 2, peerings: 1, peers: 1,
ipv4Routes: 2, ipv6Routes: 1, liveRoutes: 3,
})
// A withdrawal removes one route.
if err := db.DeleteLiveRouteBatch([]LiveRouteDeletion{
{Prefix: "203.0.113.0/24", OriginASN: 64501, PeerIP: "192.0.2.1", IPVersion: ipVersionV4},
}); err != nil {
t.Fatalf("DeleteLiveRouteBatch: %v", err)
}
stats, err = db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("GetStatsContext: %v", err)
}
assertCounts(t, "after delete", stats, wantCounts{
asns: 2, prefixes: 2, peerings: 1, peers: 1,
ipv4Routes: 1, ipv6Routes: 1, liveRoutes: 2,
})
}
// TestLiveCountsSeededFromDatabaseAtStartup writes rows, reopens the same
// database file, and checks the counts come back from the seed scan rather than
// starting at zero.
func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) {
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
if err := db.GetOrCreateASNBatch(map[int]time.Time{64500: ts, 64501: ts, 64502: ts}); err != nil {
t.Fatalf("GetOrCreateASNBatch: %v", err)
}
if err := db.UpsertLiveRouteBatch([]*LiveRoute{
mkV4Route(t, "198.51.100.0/24", 64500, ts),
mkV6Route("2001:db8::/32", 64502, ts),
}); err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
if err := db.Close(); err != nil {
t.Fatalf("Close: %v", err)
}
reopened, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to reopen database: %v", err)
}
defer func() { _ = reopened.Close() }()
stats, err := reopened.GetStatsContext(context.Background())
if err != nil {
t.Fatalf("GetStatsContext after reopen: %v", err)
}
if stats.ASNs != 3 {
t.Errorf("seeded ASNs = %d, want 3", stats.ASNs)
}
if stats.IPv4Routes != 1 || stats.IPv6Routes != 1 || stats.LiveRoutes != 2 {
t.Errorf("seeded routes = (v4 %d, v6 %d, total %d), want (1, 1, 2)",
stats.IPv4Routes, stats.IPv6Routes, stats.LiveRoutes)
}
}
// TestStatsRouteTimestamps checks the oldest/newest route timestamps are read
// from the right rows across both tables and parse into time.Time. The old
// MIN/MAX union query read its result into *time.Time, which the driver could
// not parse, so it logged a warning every call and left both timestamps nil.
func TestStatsRouteTimestamps(t *testing.T) {
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ctx := context.Background()
empty, err := db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("GetStatsContext on empty database: %v", err)
}
if empty.OldestRoute != nil || empty.NewestRoute != nil {
t.Fatalf("empty database timestamps = (%v, %v), want (nil, nil)",
empty.OldestRoute, empty.NewestRoute)
}
base := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
oldest := base
newest := base.Add(2 * time.Minute)
if err := db.UpsertLiveRouteBatch([]*LiveRoute{
mkV4Route(t, "198.51.100.0/24", 64500, base.Add(time.Minute)),
mkV4Route(t, "203.0.113.0/24", 64501, oldest),
mkV6Route("2001:db8::/32", 64502, newest),
}); err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
stats, err := db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("GetStatsContext: %v", err)
}
if stats.OldestRoute == nil || !stats.OldestRoute.Equal(oldest) {
t.Errorf("OldestRoute = %v, want %v", stats.OldestRoute, oldest)
}
if stats.NewestRoute == nil || !stats.NewestRoute.Equal(newest) {
t.Errorf("NewestRoute = %v, want %v", stats.NewestRoute, newest)
}
}
// TestLiveCountsConcurrentReadWrite runs writers and stats readers at once so
// the race detector proves the counters are safe under concurrent use.
func TestLiveCountsConcurrentReadWrite(t *testing.T) {
cfg := &config.Config{StateDir: t.TempDir()}
db, err := New(cfg, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
const writers = 4
var wg sync.WaitGroup
wg.Add(writers)
for w := range writers {
go func(base int) {
defer wg.Done()
for i := range 25 {
asn := 65000 + base*100 + i
route := mkV6Route("2001:db8::/32", asn, ts)
route.PeerIP = "2001:db8::" + uuid.NewString()
if err := db.UpsertLiveRoute(route); err != nil {
t.Errorf("UpsertLiveRoute: %v", err)
return
}
}
}(w)
}
var readerWG sync.WaitGroup
readerWG.Add(1)
stop := make(chan struct{})
go func() {
defer readerWG.Done()
for {
select {
case <-stop:
return
default:
if _, err := db.GetStatsContext(context.Background()); err != nil {
t.Errorf("GetStatsContext: %v", err)
return
}
}
}
}()
wg.Wait()
close(stop)
readerWG.Wait()
stats, err := db.GetStatsContext(context.Background())
if err != nil {
t.Fatalf("GetStatsContext: %v", err)
}
if want := writers * 25; stats.IPv6Routes != want {
t.Errorf("IPv6Routes = %d, want %d", stats.IPv6Routes, want)
}
}
type wantCounts struct {
asns int
prefixes int
peerings int
peers int
ipv4Routes int
ipv6Routes int
liveRoutes int
}
func assertCounts(t *testing.T, when string, got Stats, want wantCounts) {
t.Helper()
if got.ASNs != want.asns {
t.Errorf("%s: ASNs = %d, want %d", when, got.ASNs, want.asns)
}
if got.Prefixes != want.prefixes {
t.Errorf("%s: Prefixes = %d, want %d", when, got.Prefixes, want.prefixes)
}
if got.Peerings != want.peerings {
t.Errorf("%s: Peerings = %d, want %d", when, got.Peerings, want.peerings)
}
if got.Peers != want.peers {
t.Errorf("%s: Peers = %d, want %d", when, got.Peers, want.peers)
}
if got.IPv4Routes != want.ipv4Routes {
t.Errorf("%s: IPv4Routes = %d, want %d", when, got.IPv4Routes, want.ipv4Routes)
}
if got.IPv6Routes != want.ipv6Routes {
t.Errorf("%s: IPv6Routes = %d, want %d", when, got.IPv6Routes, want.ipv6Routes)
}
if got.LiveRoutes != want.liveRoutes {
t.Errorf("%s: LiveRoutes = %d, want %d", when, got.LiveRoutes, want.liveRoutes)
}
}
+321 -204
View File
@@ -72,6 +72,7 @@ type Database struct {
mu sync.Mutex
lockedAt time.Time
lockedBy string
counts *liveCounts
}
// New creates a new database connection and initializes the schema.
@@ -115,12 +116,19 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
db.SetMaxIdleConns(maxConns)
db.SetConnMaxLifetime(0)
database := &Database{db: db, logger: logger, path: dbPath}
database := &Database{db: db, logger: logger, path: dbPath, counts: &liveCounts{}}
if err := database.Initialize(); 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.
if err := database.seedCounts(context.Background()); err != nil {
return nil, fmt.Errorf("failed to seed statistics counters: %w", err)
}
return database, nil
}
@@ -208,6 +216,81 @@ func (d *Database) beginTx() (*loggingTx, error) {
return &loggingTx{Tx: tx, logger: d.logger}, nil
}
// A live-route upsert is an UPDATE followed, only when no row matched, by an
// INSERT. 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 two statements. The id column is set only on INSERT, so an
// updated route keeps its original id, exactly as the previous ON CONFLICT
// upsert did.
const (
updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?,
last_updated = ?, ip_start = ?, ip_end = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip,
as_path, next_hop, last_updated, ip_start, ip_end) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?,
last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip,
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
)
// upsertRouteRowV4 updates an IPv4 live route, inserting it when no row matched,
// and reports whether a new row was inserted.
func upsertRouteRowV4(upd, ins *sql.Stmt, route *LiveRoute, pathJSON string) (inserted bool, err error) {
if route.V4IPStart == nil || route.V4IPEnd == nil {
return false, fmt.Errorf("IPv4 route %s missing range values", route.Prefix)
}
res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
*route.V4IPStart, *route.V4IPEnd, route.Prefix, route.OriginASN, route.PeerIP)
if err != nil {
return false, err
}
affected, err := res.RowsAffected()
if err != nil {
return false, err
}
if affected > 0 {
return false, nil
}
_, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN,
route.PeerIP, pathJSON, route.NextHop, route.LastUpdated, *route.V4IPStart, *route.V4IPEnd)
if err != nil {
return false, err
}
return true, nil
}
// upsertRouteRowV6 updates an IPv6 live route, inserting it when no row matched,
// and reports whether a new row was inserted.
func upsertRouteRowV6(upd, ins *sql.Stmt, route *LiveRoute, pathJSON string) (inserted bool, err error) {
res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
route.Prefix, route.OriginASN, route.PeerIP)
if err != nil {
return false, err
}
affected, err := res.RowsAffected()
if err != nil {
return false, err
}
if affected > 0 {
return false, nil
}
_, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN,
route.PeerIP, pathJSON, route.NextHop, route.LastUpdated)
if err != nil {
return false, err
}
return true, nil
}
// UpsertLiveRouteBatch inserts or updates multiple live routes in a single transaction
func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
if len(routes) == 0 {
@@ -227,91 +310,65 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
}
}()
// Prepare statements for both IPv4 and IPv6
queryV4 := `
INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated, ip_start, ip_end)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
mask_length = excluded.mask_length,
as_path = excluded.as_path,
next_hop = excluded.next_hop,
last_updated = excluded.last_updated,
ip_start = excluded.ip_start,
ip_end = excluded.ip_end
`
queryV6 := `
INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
mask_length = excluded.mask_length,
as_path = excluded.as_path,
next_hop = excluded.next_hop,
last_updated = excluded.last_updated
`
stmtV4, err := tx.Prepare(queryV4)
// Prepare the update and insert statements for both tables.
updV4, err := tx.Prepare(updateLiveRouteV4SQL)
if err != nil {
return fmt.Errorf("failed to prepare IPv4 statement: %w", err)
return fmt.Errorf("failed to prepare IPv4 update statement: %w", err)
}
defer func() { _ = stmtV4.Close() }()
defer func() { _ = updV4.Close() }()
stmtV6, err := tx.Prepare(queryV6)
insV4, err := tx.Prepare(insertLiveRouteV4SQL)
if err != nil {
return fmt.Errorf("failed to prepare IPv6 statement: %w", err)
return fmt.Errorf("failed to prepare IPv4 insert statement: %w", err)
}
defer func() { _ = stmtV6.Close() }()
defer func() { _ = insV4.Close() }()
updV6, err := tx.Prepare(updateLiveRouteV6SQL)
if err != nil {
return fmt.Errorf("failed to prepare IPv6 update statement: %w", err)
}
defer func() { _ = updV6.Close() }()
insV6, err := tx.Prepare(insertLiveRouteV6SQL)
if err != nil {
return fmt.Errorf("failed to prepare IPv6 insert statement: %w", err)
}
defer func() { _ = insV6.Close() }()
var newV4, newV6 int
for _, route := range routes {
// Encode AS path as JSON
pathJSON, err := json.Marshal(route.ASPath)
if err != nil {
return fmt.Errorf("failed to encode AS path: %w", err)
}
// Use appropriate statement based on IP version
if route.IPVersion == ipVersionV4 {
// IPv4 routes must have range values
if route.V4IPStart == nil || route.V4IPEnd == nil {
return fmt.Errorf("IPv4 route %s missing range values", route.Prefix)
inserted, err := upsertRouteRowV4(updV4, insV4, route, string(pathJSON))
if err != nil {
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
}
if inserted {
newV4++
}
_, err = stmtV4.Exec(
route.ID.String(),
route.Prefix,
route.MaskLength,
route.OriginASN,
route.PeerIP,
string(pathJSON),
route.NextHop,
route.LastUpdated,
*route.V4IPStart,
*route.V4IPEnd,
)
} else {
// IPv6 routes
_, err = stmtV6.Exec(
route.ID.String(),
route.Prefix,
route.MaskLength,
route.OriginASN,
route.PeerIP,
string(pathJSON),
route.NextHop,
route.LastUpdated,
)
continue
}
inserted, err := upsertRouteRowV6(updV6, insV6, route, string(pathJSON))
if err != nil {
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
}
if inserted {
newV6++
}
}
if err = tx.Commit(); err != nil {
return fmt.Errorf("failed to commit transaction: %w", err)
}
d.counts.addRoutes(newV4, newV6)
return nil
}
@@ -362,6 +419,7 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
defer func() { _ = stmtV6WithoutOrigin.Close() }()
// Process deletions
var deletedV4, deletedV6 int64
for _, del := range deletions {
var stmt *sql.Stmt
@@ -382,21 +440,36 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
}
// Execute deletion
var res sql.Result
if del.OriginASN == 0 {
_, err = stmt.Exec(del.Prefix, del.PeerIP)
res, err = stmt.Exec(del.Prefix, del.PeerIP)
} else {
_, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP)
res, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP)
}
if err != nil {
return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err)
}
// A deletion with no origin ASN can remove several rows, so use the
// exact rows-affected count to keep the in-memory route counters right.
affected, err := res.RowsAffected()
if err != nil {
return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err)
}
if del.IPVersion == ipVersionV4 {
deletedV4 += affected
} else {
deletedV6 += affected
}
}
if err = tx.Commit(); err != nil {
return fmt.Errorf("failed to commit transaction: %w", err)
}
d.counts.addRoutes(-int(deletedV4), -int(deletedV6))
return nil
}
@@ -456,6 +529,7 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
}
defer func() { _ = insertV6Stmt.Close() }()
var newV4, newV6 int
for prefix, timestamp := range prefixes {
ipVersion := detectIPVersion(prefix)
@@ -482,6 +556,11 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
if err != nil {
return fmt.Errorf("failed to insert prefix %s: %w", prefix, err)
}
if ipVersion == ipVersionV4 {
newV4++
} else {
newV6++
}
default:
return fmt.Errorf("failed to query prefix %s: %w", prefix, err)
}
@@ -491,6 +570,8 @@ func (d *Database) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
return fmt.Errorf("failed to commit transaction: %w", err)
}
d.counts.addPrefixes(newV4, newV6)
return nil
}
@@ -534,6 +615,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
}
defer func() { _ = insertStmt.Close() }()
var newASNs int
for number, timestamp := range asns {
var asn ASN
var handle, description sql.NullString
@@ -568,6 +650,7 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
if err != nil {
return fmt.Errorf("failed to insert ASN %d: %w", number, err)
}
newASNs++
continue
}
@@ -581,6 +664,8 @@ func (d *Database) GetOrCreateASNBatch(asns map[int]time.Time) error {
return fmt.Errorf("failed to commit transaction: %w", err)
}
d.counts.addASNs(newASNs)
return nil
}
@@ -652,6 +737,8 @@ func (d *Database) GetOrCreateASN(number int, timestamp time.Time) (*ASN, error)
return nil, err
}
d.counts.addASNs(1)
return &asn, nil
}
@@ -727,6 +814,12 @@ func (d *Database) GetOrCreatePrefix(prefix string, timestamp time.Time) (*Prefi
return nil, err
}
if ipVersion == ipVersionV4 {
d.counts.addPrefixes(1, 0)
} else {
d.counts.addPrefixes(0, 1)
}
return &p, nil
}
@@ -803,6 +896,10 @@ func (d *Database) RecordPeering(asA, asB int, timestamp time.Time) error {
return err
}
if !exists {
d.counts.addPeerings(1)
}
return nil
}
@@ -846,6 +943,7 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error {
}
defer func() { _ = insertStmt.Close() }()
var newPeers int
for _, update := range peers {
var exists bool
err = checkStmt.QueryRow(update.PeerIP).Scan(&exists)
@@ -863,12 +961,17 @@ func (d *Database) UpdatePeerBatch(peers map[string]PeerUpdate) error {
if err != nil {
return fmt.Errorf("failed to update peer %s: %w", update.PeerIP, err)
}
if !exists {
newPeers++
}
}
if err = tx.Commit(); err != nil {
return fmt.Errorf("failed to commit transaction: %w", err)
}
d.counts.addPeers(newPeers)
return nil
}
@@ -918,6 +1021,10 @@ func (d *Database) UpdatePeer(peerIP string, peerASN int, messageType string, ti
return err
}
if !exists {
d.counts.addPeers(1)
}
return nil
}
@@ -926,169 +1033,161 @@ func (d *Database) GetStats() (Stats, error) {
return d.GetStatsContext(context.Background())
}
// GetStatsContext returns database statistics with context support
// GetStatsContext returns database statistics with context support.
//
// The row counts (ASNs, prefixes, peerings, peers, live routes) come from the
// in-memory counters, seeded at startup and kept current on every write, so a
// read runs no COUNT(*) over the tables. The oldest/newest route timestamps are
// read from the ends of the last_updated index, and the file size from a
// stat(); neither is a table scan. The only remaining query is the prefix
// distribution.
func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) {
var stats Stats
// Count ASNs
err := d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM asns").Scan(&stats.ASNs)
if err != nil {
return stats, err
}
// Row counts from memory, as a single consistent snapshot.
d.counts.fill(&stats)
// Count prefixes from both tables
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM prefixes_v4").Scan(&stats.IPv4Prefixes)
if err != nil {
return stats, err
}
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM prefixes_v6").Scan(&stats.IPv6Prefixes)
if err != nil {
return stats, err
}
stats.Prefixes = stats.IPv4Prefixes + stats.IPv6Prefixes
// Count peerings
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM peerings").Scan(&stats.Peerings)
if err != nil {
return stats, err
}
// Count peers
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM bgp_peers").Scan(&stats.Peers)
if err != nil {
return stats, err
}
// Get database file size
fileInfo, err := os.Stat(d.path)
if err != nil {
// Database file size is a cheap stat() on the file.
if fileInfo, err := os.Stat(d.path); err != nil {
d.logger.Warn("Failed to get database file size", "error", err)
stats.FileSizeBytes = 0
} else {
stats.FileSizeBytes = fileInfo.Size()
}
// Get live routes count from both tables
var v4Count, v6Count int
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v4").Scan(&v4Count)
if err != nil {
return stats, fmt.Errorf("failed to count IPv4 routes: %w", err)
}
err = d.db.QueryRowContext(ctx, "SELECT COUNT(*) FROM live_routes_v6").Scan(&v6Count)
if err != nil {
return stats, fmt.Errorf("failed to count IPv6 routes: %w", err)
}
stats.LiveRoutes = v4Count + v6Count
// Get oldest and newest route timestamps
routeTimestampQuery := `
SELECT MIN(last_updated), MAX(last_updated) FROM (
SELECT last_updated FROM live_routes_v4
UNION ALL
SELECT last_updated FROM live_routes_v6
)
`
var oldestRoute, newestRoute *time.Time
err = d.db.QueryRowContext(ctx, routeTimestampQuery).Scan(&oldestRoute, &newestRoute)
// 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 = oldestRoute
stats.NewestRoute = newestRoute
stats.OldestRoute = oldest
stats.NewestRoute = newest
}
// Get prefix distribution
// Prefix distribution counts distinct prefixes per mask length. It stays a
// query over the covering (mask_length, prefix) index rather than an
// in-memory counter: maintaining distinct-prefix-per-mask in memory would
// need a per-prefix table of roughly a million entries, memory this service
// is tuned to avoid.
stats.IPv4PrefixDistribution, stats.IPv6PrefixDistribution, err = d.GetPrefixDistributionContext(ctx)
if err != nil {
// Log but don't fail
// Log but don't fail.
d.logger.Warn("Failed to get prefix distribution", "error", err)
}
return stats, 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) (oldest, newest *time.Time, err error) {
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. ok 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) (ts time.Time, ok bool, err error) {
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
}
}
// UpsertLiveRoute inserts or updates a live route
func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
d.lock("UpsertLiveRoute")
defer d.unlock()
// Choose table based on IP version
tableName := "live_routes_v4"
if route.IPVersion == ipVersionV6 {
tableName = "live_routes_v6"
}
var query string
if route.IPVersion == ipVersionV4 {
query = fmt.Sprintf(`
INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated, ip_start, ip_end)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
mask_length = excluded.mask_length,
as_path = excluded.as_path,
next_hop = excluded.next_hop,
last_updated = excluded.last_updated,
ip_start = excluded.ip_start,
ip_end = excluded.ip_end
`, tableName)
} else {
query = fmt.Sprintf(`
INSERT INTO %s (id, prefix, mask_length, origin_asn, peer_ip, as_path, next_hop,
last_updated)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(prefix, origin_asn, peer_ip) DO UPDATE SET
mask_length = excluded.mask_length,
as_path = excluded.as_path,
next_hop = excluded.next_hop,
last_updated = excluded.last_updated
`, tableName)
}
// Encode AS path as JSON
pathJSON, err := json.Marshal(route.ASPath)
if err != nil {
return fmt.Errorf("failed to encode AS path: %w", err)
}
if route.IPVersion == ipVersionV4 {
// Convert v4_ip_start and v4_ip_end to interface{} for SQL NULL handling
var v4Start, v4End interface{}
if route.V4IPStart != nil {
v4Start = *route.V4IPStart
}
if route.V4IPEnd != nil {
v4End = *route.V4IPEnd
}
_, err = d.db.Exec(query,
route.ID.String(),
route.Prefix,
route.MaskLength,
route.OriginASN,
route.PeerIP,
string(pathJSON),
route.NextHop,
route.LastUpdated,
v4Start,
v4End,
)
} else {
_, err = d.db.Exec(query,
route.ID.String(),
route.Prefix,
route.MaskLength,
route.OriginASN,
route.PeerIP,
string(pathJSON),
route.NextHop,
route.LastUpdated,
)
updateSQL, insertSQL := updateLiveRouteV4SQL, insertLiveRouteV4SQL
if route.IPVersion == ipVersionV6 {
updateSQL, insertSQL = updateLiveRouteV6SQL, insertLiveRouteV6SQL
}
return err
// 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.
upd, err := d.db.Prepare(updateSQL)
if err != nil {
return fmt.Errorf("failed to prepare update statement: %w", err)
}
defer func() { _ = upd.Close() }()
ins, err := d.db.Prepare(insertSQL)
if err != nil {
return fmt.Errorf("failed to prepare insert statement: %w", err)
}
defer func() { _ = ins.Close() }()
var inserted bool
if route.IPVersion == ipVersionV4 {
inserted, err = upsertRouteRowV4(upd, ins, route, string(pathJSON))
} else {
inserted, err = upsertRouteRowV6(upd, ins, 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)
}
}
return nil
}
// DeleteLiveRoute deletes a live route
@@ -1103,23 +1202,41 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string)
return fmt.Errorf("invalid prefix format: %w", err)
}
tableName := "live_routes_v4"
if ipnet.IP.To4() == nil {
tableName = "live_routes_v6"
isV4 := ipnet.IP.To4() != nil
// Literal per-table queries (rather than one formatted with the table name)
// so the delete carries no dynamically built SQL. A delete with no origin
// ASN can remove several rows.
var res sql.Result
switch {
case isV4 && originASN == 0:
res, err = d.db.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
case isV4:
res, err = d.db.Exec(
`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
prefix, originASN, peerIP)
case originASN == 0:
res, err = d.db.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
default:
res, err = d.db.Exec(
`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
prefix, originASN, peerIP)
}
if err != nil {
return err
}
var query string
if originASN == 0 {
// Delete all routes for this prefix from this peer
query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND peer_ip = ?`, tableName)
_, err = d.db.Exec(query, prefix, peerIP)
affected, err := res.RowsAffected()
if err != nil {
return err
}
if isV4 {
d.counts.addRoutes(-int(affected), 0)
} else {
// Delete specific route
query = fmt.Sprintf(`DELETE FROM %s WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`, tableName)
_, err = d.db.Exec(query, prefix, originASN, peerIP)
d.counts.addRoutes(0, -int(affected))
}
return err
return nil
}
// GetPrefixDistribution returns the distribution of unique prefixes by mask length
+2
View File
@@ -18,6 +18,8 @@ type Stats struct {
Peers int
FileSizeBytes int64
LiveRoutes int
IPv4Routes int
IPv6Routes int
OldestRoute *time.Time
NewestRoute *time.Time
IPv4PrefixDistribution []PrefixDistribution
+4 -18
View File
@@ -219,13 +219,6 @@ func (s *Server) handleStatusJSON() http.HandlerFunc {
const bitsPerMegabit = 1000000.0
// Get route counts from database
ipv4Routes, ipv6Routes, err := s.db.GetLiveRouteCountsContext(ctx)
if err != nil {
s.logger.Warn("Failed to get live route counts", "error", err)
// Continue with zero counts
}
// Get route update metrics
routeMetrics := s.streamer.GetMetricsTracker().GetRouteMetrics()
@@ -259,8 +252,8 @@ func (s *Server) handleStatusJSON() http.HandlerFunc {
Peers: dbStats.Peers,
DatabaseSizeBytes: dbStats.FileSizeBytes,
LiveRoutes: dbStats.LiveRoutes,
IPv4Routes: ipv4Routes,
IPv6Routes: ipv6Routes,
IPv4Routes: dbStats.IPv4Routes,
IPv6Routes: dbStats.IPv6Routes,
OldestRoute: dbStats.OldestRoute,
NewestRoute: dbStats.NewestRoute,
IPv4UpdatesPerSec: routeMetrics.IPv4UpdatesPerSec,
@@ -439,13 +432,6 @@ func (s *Server) handleStats() http.HandlerFunc {
const bitsPerMegabit = 1000000.0
// Get route counts from database
ipv4Routes, ipv6Routes, err := s.db.GetLiveRouteCountsContext(ctx)
if err != nil {
s.logger.Warn("Failed to get live route counts", "error", err)
// Continue with zero counts
}
// Get route update metrics
routeMetrics := s.streamer.GetMetricsTracker().GetRouteMetrics()
@@ -537,8 +523,8 @@ func (s *Server) handleStats() http.HandlerFunc {
Peers: dbStats.Peers,
DatabaseSizeBytes: dbStats.FileSizeBytes,
LiveRoutes: dbStats.LiveRoutes,
IPv4Routes: ipv4Routes,
IPv6Routes: ipv6Routes,
IPv4Routes: dbStats.IPv4Routes,
IPv6Routes: dbStats.IPv6Routes,
OldestRoute: dbStats.OldestRoute,
NewestRoute: dbStats.NewestRoute,
IPv4UpdatesPerSec: routeMetrics.IPv4UpdatesPerSec,