Compare commits
3
Commits
3e35fae822
...
next
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a9b68608c6 | ||
|
|
44a5f4cdca | ||
|
|
32704c601e |
@@ -142,7 +142,10 @@ routewatch/
|
|||||||
- **Backpressure**: Probabilistic message dropping when queues exceed 50% capacity
|
- **Backpressure**: Probabilistic message dropping when queues exceed 50% capacity
|
||||||
- **Graceful Shutdown**: 60-second timeout, flushes all pending batches
|
- **Graceful Shutdown**: 60-second timeout, flushes all pending batches
|
||||||
- **Reconnection**: Exponential backoff (5s-320s) with reset after 30s of stable connection
|
- **Reconnection**: Exponential backoff (5s-320s) with reset after 30s of stable connection
|
||||||
- **IPv4 Optimization**: IP ranges stored as uint32 for O(1) lookups
|
- **IP Lookup**: the most specific live route is found by looking up the
|
||||||
|
address's prefix at each mask length, longest first, in the prefix index
|
||||||
|
(at most 33 lookups for IPv4, 129 for IPv6); prefixes are stored in the
|
||||||
|
text form Go's `net/netip` prints
|
||||||
|
|
||||||
### Database Schema
|
### Database Schema
|
||||||
|
|
||||||
@@ -154,7 +157,7 @@ prefixes_v6(id, prefix, mask_length, first_seen, last_seen)
|
|||||||
|
|
||||||
-- Live routing tables (one per IP version)
|
-- Live routing tables (one per IP version)
|
||||||
live_routes_v4(id, prefix, mask_length, origin_asn, peer_ip, as_path,
|
live_routes_v4(id, prefix, mask_length, origin_asn, peer_ip, as_path,
|
||||||
next_hop, last_updated, v4_ip_start, v4_ip_end)
|
next_hop, last_updated)
|
||||||
live_routes_v6(id, prefix, mask_length, origin_asn, peer_ip, as_path,
|
live_routes_v6(id, prefix, mask_length, origin_asn, peer_ip, as_path,
|
||||||
next_hop, last_updated)
|
next_hop, last_updated)
|
||||||
|
|
||||||
|
|||||||
@@ -24,10 +24,22 @@ https://git.eeqj.de/sneak/routewatch/pulls/6. After that, setting
|
|||||||
routewatch up under upaas on fsn1app1 and deploying it are his
|
routewatch up under upaas on fsn1app1 and deploying it are his
|
||||||
(https://git.eeqj.de/sneak/routewatch/issues/31), and so is the run under a
|
(https://git.eeqj.de/sneak/routewatch/issues/31), and so is the run under a
|
||||||
real 5 GiB limit (https://git.eeqj.de/sneak/routewatch/issues/3).
|
real 5 GiB limit (https://git.eeqj.de/sneak/routewatch/issues/3).
|
||||||
The other open issue is https://git.eeqj.de/sneak/routewatch/issues/30.
|
|
||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
|
- 2026-10-03: `/api/v1/stats` serves the prefix distribution from memory,
|
||||||
|
seeded at startup and adjusted on every live-route write, so a request no
|
||||||
|
longer reads every live route (closes #30)
|
||||||
|
- 2026-10-03: a new database is created with `auto_vacuum` set to
|
||||||
|
incremental, through the connection string so it is set before the file
|
||||||
|
is first written, and each periodic incremental vacuum now returns up to
|
||||||
|
1000 free pages; the late `PRAGMA auto_vacuum` in `Initialize`, which
|
||||||
|
SQLite ignored, is gone (closes #43)
|
||||||
|
- 2026-10-03: looking up an IP address no longer reads every IPv6 route:
|
||||||
|
both families find the most specific live route with at most 33 or 129
|
||||||
|
lookups on the prefix index, and the IPv4 range columns are gone. Prefixes
|
||||||
|
from the feed are stored in one text form, so an IPv6 withdrawal, which the
|
||||||
|
feed sends uncompressed, now removes its route (closes #48)
|
||||||
- 2026-10-02: a plain `docker build .` stamps the commit's tag or short
|
- 2026-10-02: a plain `docker build .` stamps the commit's tag or short
|
||||||
commit (`git describe --tags --always`) into the page footer instead of
|
commit (`git describe --tags --always`) into the page footer instead of
|
||||||
`unknown`: `.dockerignore` sends `.git` without `.git/config`, a `VERSION`
|
`unknown`: `.dockerignore` sends `.git` without `.git/config`, a `VERSION`
|
||||||
@@ -108,6 +120,3 @@ The other open issue is https://git.eeqj.de/sneak/routewatch/issues/30.
|
|||||||
- Production memory under 5 GiB: whether to test under a real 5 GiB
|
- Production memory under 5 GiB: whether to test under a real 5 GiB
|
||||||
container limit on fsn1app1 is open for sneak
|
container limit on fsn1app1 is open for sneak
|
||||||
(https://git.eeqj.de/sneak/routewatch/issues/3)
|
(https://git.eeqj.de/sneak/routewatch/issues/3)
|
||||||
- `/api/v1/stats` answered HTTP 500 after 35 hours on the live feed, seen
|
|
||||||
on `3898daa`, which predates the 2026-09-22 in-memory statistics
|
|
||||||
(https://git.eeqj.de/sneak/routewatch/issues/30)
|
|
||||||
|
|||||||
@@ -6,11 +6,12 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
)
|
)
|
||||||
|
|
||||||
// liveCounts holds the running row counts that the stats endpoints report. They
|
// liveCounts holds the running row counts and the prefix distribution that the
|
||||||
// are seeded once at startup from the tables and then adjusted on every write,
|
// stats endpoints report. They are seeded once at startup from the tables and
|
||||||
// so a stats read serves them from memory instead of running a COUNT(*) over
|
// then adjusted on every write, so a stats read serves them from memory instead
|
||||||
// each table. Those scans, once the database passed a few GiB, took the whole
|
// of running a query over the tables. The COUNT(*) scans (issue 27) and then
|
||||||
// request timeout and made /api/v1/stats return 500 (issue 27).
|
// the prefix distribution query (issue 30) each grew with the database until
|
||||||
|
// they took the whole request timeout and made /api/v1/stats return 500.
|
||||||
//
|
//
|
||||||
// A single mutex guards all fields so the stats reader takes a consistent
|
// 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
|
// snapshot at one instant and writers, which already run under the database
|
||||||
@@ -24,11 +25,16 @@ type liveCounts struct {
|
|||||||
peers int
|
peers int
|
||||||
routesV4 int
|
routesV4 int
|
||||||
routesV6 int
|
routesV6 int
|
||||||
|
// The prefix distribution: for each mask length, the number of distinct
|
||||||
|
// prefixes that have at least one live route.
|
||||||
|
distributionV4 [ipv4Bits + 1]int
|
||||||
|
distributionV6 [ipv6Bits + 1]int
|
||||||
}
|
}
|
||||||
|
|
||||||
// seed sets every count to the value read from the tables at startup. It runs
|
// 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.
|
// before any writer, so it needs no coordination with the adjust methods.
|
||||||
func (c *liveCounts) seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6 int) {
|
func (c *liveCounts) seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6 int,
|
||||||
|
distributionV4, distributionV6 []PrefixDistribution) {
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
defer c.mu.Unlock()
|
defer c.mu.Unlock()
|
||||||
|
|
||||||
@@ -39,6 +45,12 @@ func (c *liveCounts) seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV
|
|||||||
c.peers = peers
|
c.peers = peers
|
||||||
c.routesV4 = routesV4
|
c.routesV4 = routesV4
|
||||||
c.routesV6 = routesV6
|
c.routesV6 = routesV6
|
||||||
|
for _, entry := range distributionV4 {
|
||||||
|
addAtMaskLength(c.distributionV4[:], entry.MaskLength, entry.Count)
|
||||||
|
}
|
||||||
|
for _, entry := range distributionV6 {
|
||||||
|
addAtMaskLength(c.distributionV6[:], entry.MaskLength, entry.Count)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// addASNs adds n to the ASN count.
|
// addASNs adds n to the ASN count.
|
||||||
@@ -79,6 +91,30 @@ func (c *liveCounts) addRoutes(v4, v6 int) {
|
|||||||
c.mu.Unlock()
|
c.mu.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// addToDistribution adds n to the IPv4 and IPv6 prefix distributions once for
|
||||||
|
// each listed mask length. A write lists the mask lengths of the prefixes it
|
||||||
|
// gave their first live route with n = 1, and of the prefixes it left with no
|
||||||
|
// live route with n = -1.
|
||||||
|
func (c *liveCounts) addToDistribution(maskLengthsV4, maskLengthsV6 []int, n int) {
|
||||||
|
c.mu.Lock()
|
||||||
|
defer c.mu.Unlock()
|
||||||
|
|
||||||
|
for _, maskLength := range maskLengthsV4 {
|
||||||
|
addAtMaskLength(c.distributionV4[:], maskLength, n)
|
||||||
|
}
|
||||||
|
for _, maskLength := range maskLengthsV6 {
|
||||||
|
addAtMaskLength(c.distributionV6[:], maskLength, n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// addAtMaskLength adds n to counts[maskLength]. A mask length the array has no
|
||||||
|
// entry for is ignored, so a malformed route cannot crash the daemon.
|
||||||
|
func addAtMaskLength(counts []int, maskLength, n int) {
|
||||||
|
if maskLength >= 0 && maskLength < len(counts) {
|
||||||
|
counts[maskLength] += n
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// fill copies the counts into a Stats, including the derived totals, under a
|
// fill copies the counts into a Stats, including the derived totals, under a
|
||||||
// single read lock so the reader sees one consistent snapshot.
|
// single read lock so the reader sees one consistent snapshot.
|
||||||
func (c *liveCounts) fill(s *Stats) {
|
func (c *liveCounts) fill(s *Stats) {
|
||||||
@@ -94,6 +130,21 @@ func (c *liveCounts) fill(s *Stats) {
|
|||||||
s.IPv4Routes = c.routesV4
|
s.IPv4Routes = c.routesV4
|
||||||
s.IPv6Routes = c.routesV6
|
s.IPv6Routes = c.routesV6
|
||||||
s.LiveRoutes = c.routesV4 + c.routesV6
|
s.LiveRoutes = c.routesV4 + c.routesV6
|
||||||
|
s.IPv4PrefixDistribution = distributionList(c.distributionV4[:])
|
||||||
|
s.IPv6PrefixDistribution = distributionList(c.distributionV6[:])
|
||||||
|
}
|
||||||
|
|
||||||
|
// distributionList lists the mask lengths that have at least one prefix, in
|
||||||
|
// ascending order, the way the distribution query returns them.
|
||||||
|
func distributionList(counts []int) []PrefixDistribution {
|
||||||
|
var list []PrefixDistribution
|
||||||
|
for maskLength, count := range counts {
|
||||||
|
if count > 0 {
|
||||||
|
list = append(list, PrefixDistribution{MaskLength: maskLength, Count: count})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return list
|
||||||
}
|
}
|
||||||
|
|
||||||
// countRows returns the number of rows in the named table. It is used only at
|
// countRows returns the number of rows in the named table. It is used only at
|
||||||
@@ -108,8 +159,9 @@ func (d *Database) countRows(ctx context.Context, table string) (int, error) {
|
|||||||
return n, nil
|
return n, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// seedCounts reads the current row counts from the tables into the in-memory
|
// seedCounts reads the current row counts and prefix distribution from the
|
||||||
// counters. It runs once at startup, before the streamer begins writing.
|
// tables into the in-memory counters. It runs once at startup, before the
|
||||||
|
// streamer begins writing.
|
||||||
func (d *Database) seedCounts(ctx context.Context) error {
|
func (d *Database) seedCounts(ctx context.Context) error {
|
||||||
asns, err := d.countRows(ctx, "asns")
|
asns, err := d.countRows(ctx, "asns")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -139,8 +191,13 @@ func (d *Database) seedCounts(ctx context.Context) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
distributionV4, distributionV6, err := d.GetPrefixDistributionContext(ctx)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
d.counts.seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6)
|
d.counts.seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6,
|
||||||
|
distributionV4, distributionV6)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package database
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"slices"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -11,27 +12,26 @@ import (
|
|||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
)
|
)
|
||||||
|
|
||||||
// mkV4Route builds an IPv4 live route with its range columns populated.
|
// mkV4Route builds an IPv4 live route with its mask length taken from the
|
||||||
|
// prefix.
|
||||||
func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute {
|
func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
start, end, err := CalculateIPv4Range(prefix)
|
maskLength, err := prefixMaskLength(prefix)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("CalculateIPv4Range(%s): %v", prefix, err)
|
t.Fatalf("prefixMaskLength(%s): %v", prefix, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
return &LiveRoute{
|
return &LiveRoute{
|
||||||
ID: uuid.New(),
|
ID: uuid.New(),
|
||||||
Prefix: prefix,
|
Prefix: prefix,
|
||||||
MaskLength: 24,
|
MaskLength: maskLength,
|
||||||
IPVersion: ipVersionV4,
|
IPVersion: ipVersionV4,
|
||||||
OriginASN: asn,
|
OriginASN: asn,
|
||||||
PeerIP: "192.0.2.1",
|
PeerIP: "192.0.2.1",
|
||||||
ASPath: []int{asn},
|
ASPath: []int{asn},
|
||||||
NextHop: "192.0.2.254",
|
NextHop: "192.0.2.254",
|
||||||
LastUpdated: ts,
|
LastUpdated: ts,
|
||||||
V4IPStart: &start,
|
|
||||||
V4IPEnd: &end,
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -143,8 +143,8 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// TestLiveCountsSeededFromDatabaseAtStartup writes rows, reopens the same
|
// TestLiveCountsSeededFromDatabaseAtStartup writes rows, reopens the same
|
||||||
// database file, and checks the counts come back from the seed scan rather than
|
// database file, and checks the counts and the prefix distribution come back
|
||||||
// starting at zero.
|
// from the seed scan rather than starting at zero.
|
||||||
func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) {
|
func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) {
|
||||||
cfg := &config.Config{StateDir: t.TempDir()}
|
cfg := &config.Config{StateDir: t.TempDir()}
|
||||||
|
|
||||||
@@ -184,6 +184,171 @@ func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) {
|
|||||||
t.Errorf("seeded routes = (v4 %d, v6 %d, total %d), want (1, 1, 2)",
|
t.Errorf("seeded routes = (v4 %d, v6 %d, total %d), want (1, 1, 2)",
|
||||||
stats.IPv4Routes, stats.IPv6Routes, stats.LiveRoutes)
|
stats.IPv4Routes, stats.IPv6Routes, stats.LiveRoutes)
|
||||||
}
|
}
|
||||||
|
assertDistribution(t, "seeded IPv4 distribution", stats.IPv4PrefixDistribution,
|
||||||
|
[]PrefixDistribution{{MaskLength: 24, Count: 1}})
|
||||||
|
assertDistribution(t, "seeded IPv6 distribution", stats.IPv6PrefixDistribution,
|
||||||
|
[]PrefixDistribution{{MaskLength: 32, Count: 1}})
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestPrefixDistributionTracksWrites checks that the prefix distribution the
|
||||||
|
// stats read reports stays exact across each kind of live-route write, and that
|
||||||
|
// after every step it equals what the distribution query reads from the route
|
||||||
|
// tables. The steps run once through the batch methods the prefix handler uses
|
||||||
|
// and once through the single-route methods.
|
||||||
|
func TestPrefixDistributionTracksWrites(t *testing.T) {
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
shared := mkV4Route(t, "198.51.100.0/24", 64500, ts)
|
||||||
|
sharedSecondPeer := mkV4Route(t, "198.51.100.0/24", 64500, ts)
|
||||||
|
sharedSecondPeer.PeerIP = "192.0.2.2"
|
||||||
|
other := mkV4Route(t, "203.0.113.0/24", 64501, ts)
|
||||||
|
wide := mkV4Route(t, "172.16.0.0/16", 64502, ts)
|
||||||
|
v6 := mkV6Route("2001:db8::/32", 64503, ts)
|
||||||
|
v6SecondPeer := mkV6Route("2001:db8::/32", 64503, ts)
|
||||||
|
v6SecondPeer.PeerIP = "2001:db8::2"
|
||||||
|
|
||||||
|
all := []PrefixDistribution{{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 2}}
|
||||||
|
v6Only := []PrefixDistribution{{MaskLength: 32, Count: 1}}
|
||||||
|
|
||||||
|
steps := []struct {
|
||||||
|
name string
|
||||||
|
announce []*LiveRoute
|
||||||
|
withdraw []*LiveRoute
|
||||||
|
wantV4 []PrefixDistribution
|
||||||
|
wantV6 []PrefixDistribution
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "new routes",
|
||||||
|
announce: []*LiveRoute{shared, other, wide, v6},
|
||||||
|
wantV4: all, wantV6: v6Only,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "re-announcement",
|
||||||
|
announce: []*LiveRoute{shared, other, wide, v6},
|
||||||
|
wantV4: all, wantV6: v6Only,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "second peer announces a prefix that has a live route",
|
||||||
|
announce: []*LiveRoute{sharedSecondPeer, v6SecondPeer},
|
||||||
|
wantV4: all, wantV6: v6Only,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "withdrawal of a route that is not the last for its prefix",
|
||||||
|
withdraw: []*LiveRoute{shared, v6},
|
||||||
|
wantV4: all, wantV6: v6Only,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "withdrawal of the last route for a prefix",
|
||||||
|
withdraw: []*LiveRoute{sharedSecondPeer, v6SecondPeer},
|
||||||
|
wantV4: []PrefixDistribution{{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 1}},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "withdrawal of every remaining route",
|
||||||
|
withdraw: []*LiveRoute{other, wide},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "two peers announce a new prefix together",
|
||||||
|
announce: []*LiveRoute{shared, sharedSecondPeer, v6, v6SecondPeer},
|
||||||
|
wantV4: []PrefixDistribution{{MaskLength: 24, Count: 1}},
|
||||||
|
wantV6: v6Only,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "both routes for a prefix withdrawn together",
|
||||||
|
withdraw: []*LiveRoute{shared, sharedSecondPeer, v6, v6SecondPeer},
|
||||||
|
},
|
||||||
|
// The feed often withdraws a route that is not live. That must not take
|
||||||
|
// the prefix out of the distribution. A count wrongly taken below zero is
|
||||||
|
// left out of the answer, so the next step shows it: its announcement
|
||||||
|
// at the same mask length would then not be counted.
|
||||||
|
{
|
||||||
|
name: "withdrawal of routes that are not live",
|
||||||
|
withdraw: []*LiveRoute{other, v6},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "announcement after a withdrawal of routes that are not live",
|
||||||
|
announce: []*LiveRoute{other, v6},
|
||||||
|
wantV4: []PrefixDistribution{{MaskLength: 24, Count: 1}},
|
||||||
|
wantV6: v6Only,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, batch := range []bool{true, false} {
|
||||||
|
name := "single-route writes"
|
||||||
|
if batch {
|
||||||
|
name = "batch writes"
|
||||||
|
}
|
||||||
|
|
||||||
|
t.Run(name, func(t *testing.T) {
|
||||||
|
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to create database: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
for _, step := range steps {
|
||||||
|
if err := announceRoutes(db, batch, step.announce); err != nil {
|
||||||
|
t.Fatalf("%s: announce: %v", step.name, err)
|
||||||
|
}
|
||||||
|
if err := withdrawRoutes(db, batch, step.withdraw); err != nil {
|
||||||
|
t.Fatalf("%s: withdraw: %v", step.name, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
stats, err := db.GetStatsContext(ctx)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("%s: GetStatsContext: %v", step.name, err)
|
||||||
|
}
|
||||||
|
queryV4, queryV6, err := db.GetPrefixDistributionContext(ctx)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("%s: GetPrefixDistributionContext: %v", step.name, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
assertDistribution(t, step.name+": IPv4 distribution", stats.IPv4PrefixDistribution, step.wantV4)
|
||||||
|
assertDistribution(t, step.name+": IPv6 distribution", stats.IPv6PrefixDistribution, step.wantV6)
|
||||||
|
assertDistribution(t, step.name+": IPv4 distribution query", queryV4, step.wantV4)
|
||||||
|
assertDistribution(t, step.name+": IPv6 distribution query", queryV6, step.wantV6)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// announceRoutes writes routes in one UpsertLiveRouteBatch, or with one
|
||||||
|
// UpsertLiveRoute each.
|
||||||
|
func announceRoutes(db *Database, batch bool, routes []*LiveRoute) error {
|
||||||
|
if batch {
|
||||||
|
return db.UpsertLiveRouteBatch(routes)
|
||||||
|
}
|
||||||
|
for _, route := range routes {
|
||||||
|
if err := db.UpsertLiveRoute(route); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// withdrawRoutes removes routes in one DeleteLiveRouteBatch, or with one
|
||||||
|
// DeleteLiveRoute each. It names each route by prefix and peer only, with no
|
||||||
|
// origin ASN, as a withdrawal from the feed does when its message carries no AS
|
||||||
|
// path.
|
||||||
|
func withdrawRoutes(db *Database, batch bool, routes []*LiveRoute) error {
|
||||||
|
if !batch {
|
||||||
|
for _, route := range routes {
|
||||||
|
if err := db.DeleteLiveRoute(route.Prefix, 0, route.PeerIP); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
deletions := make([]LiveRouteDeletion, 0, len(routes))
|
||||||
|
for _, route := range routes {
|
||||||
|
deletions = append(deletions, LiveRouteDeletion{
|
||||||
|
Prefix: route.Prefix, PeerIP: route.PeerIP, IPVersion: route.IPVersion,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
return db.DeleteLiveRouteBatch(deletions)
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestStatsRouteTimestamps checks the oldest/newest route timestamps are read
|
// TestStatsRouteTimestamps checks the oldest/newest route timestamps are read
|
||||||
@@ -296,6 +461,9 @@ func TestLiveCountsConcurrentReadWrite(t *testing.T) {
|
|||||||
if want := writers * 25; stats.IPv6Routes != want {
|
if want := writers * 25; stats.IPv6Routes != want {
|
||||||
t.Errorf("IPv6Routes = %d, want %d", stats.IPv6Routes, want)
|
t.Errorf("IPv6Routes = %d, want %d", stats.IPv6Routes, want)
|
||||||
}
|
}
|
||||||
|
// Every writer announced the same prefix, so it counts once.
|
||||||
|
assertDistribution(t, "IPv6 distribution", stats.IPv6PrefixDistribution,
|
||||||
|
[]PrefixDistribution{{MaskLength: 32, Count: 1}})
|
||||||
}
|
}
|
||||||
|
|
||||||
type wantCounts struct {
|
type wantCounts struct {
|
||||||
@@ -333,3 +501,11 @@ func assertCounts(t *testing.T, when string, got Stats, want wantCounts) {
|
|||||||
t.Errorf("%s: LiveRoutes = %d, want %d", when, got.LiveRoutes, want.liveRoutes)
|
t.Errorf("%s: LiveRoutes = %d, want %d", when, got.LiveRoutes, want.liveRoutes)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func assertDistribution(t *testing.T, what string, got, want []PrefixDistribution) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
if !slices.Equal(got, want) {
|
||||||
|
t.Errorf("%s = %v, want %v", what, got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+259
-376
@@ -9,6 +9,7 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net"
|
"net"
|
||||||
|
"net/netip"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"runtime"
|
"runtime"
|
||||||
@@ -32,10 +33,8 @@ const (
|
|||||||
dirPermissions = 0750 // rwxr-x---
|
dirPermissions = 0750 // rwxr-x---
|
||||||
ipVersionV4 = 4
|
ipVersionV4 = 4
|
||||||
ipVersionV6 = 6
|
ipVersionV6 = 6
|
||||||
ipv6Length = 16
|
|
||||||
ipv4Offset = 12
|
|
||||||
ipv4Bits = 32
|
ipv4Bits = 32
|
||||||
maxIPv4 = 0xFFFFFFFF
|
ipv6Bits = 128
|
||||||
)
|
)
|
||||||
|
|
||||||
// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every
|
// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every
|
||||||
@@ -94,8 +93,13 @@ func New(cfg *config.Config, logger *logger.Logger) (*Database, error) {
|
|||||||
// a transaction that reads before writing starts as a reader and, when 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
|
// then writes while another connection holds the write lock, fails at once
|
||||||
// with "database is locked" without waiting for _busy_timeout.
|
// 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(
|
dsn := fmt.Sprintf(
|
||||||
"file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d&_journal_mode=WAL&_txlock=immediate",
|
"file:%s?_cache_size=%d&_synchronous=OFF&_busy_timeout=%d"+
|
||||||
|
"&_auto_vacuum=incremental&_journal_mode=WAL&_txlock=immediate",
|
||||||
dbPath,
|
dbPath,
|
||||||
sqliteCacheSizeKiB,
|
sqliteCacheSizeKiB,
|
||||||
sqliteBusyTimeoutMs,
|
sqliteBusyTimeoutMs,
|
||||||
@@ -142,7 +146,6 @@ func (d *Database) Initialize() error {
|
|||||||
pragmas := []string{
|
pragmas := []string{
|
||||||
"PRAGMA journal_mode=WAL", // Write-Ahead Logging
|
"PRAGMA journal_mode=WAL", // Write-Ahead Logging
|
||||||
"PRAGMA analysis_limit=0", // Disable automatic ANALYZE
|
"PRAGMA analysis_limit=0", // Disable automatic ANALYZE
|
||||||
"PRAGMA auto_vacuum=INCREMENTAL", // Enable incremental vacuum
|
|
||||||
fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes),
|
fmt.Sprintf("PRAGMA soft_heap_limit=%d", sqliteSoftHeapLimitBytes),
|
||||||
fmt.Sprintf("PRAGMA hard_heap_limit=%d", sqliteHardHeapLimitBytes),
|
fmt.Sprintf("PRAGMA hard_heap_limit=%d", sqliteHardHeapLimitBytes),
|
||||||
}
|
}
|
||||||
@@ -225,10 +228,10 @@ func (d *Database) beginTx() (*loggingTx, error) {
|
|||||||
// upsert did.
|
// upsert did.
|
||||||
const (
|
const (
|
||||||
updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?,
|
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 = ?`
|
last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
|
||||||
|
|
||||||
insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
|
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
||||||
|
|
||||||
updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?,
|
updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?,
|
||||||
last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
|
last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
|
||||||
@@ -237,58 +240,76 @@ const (
|
|||||||
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
||||||
)
|
)
|
||||||
|
|
||||||
// upsertRouteRowV4 updates an IPv4 live route, inserting it when no row matched,
|
// Before a new route is inserted and after a route is deleted, the write looks
|
||||||
// and reports whether a new row was inserted.
|
// up whether any live route has that prefix, to keep the in-memory prefix
|
||||||
func upsertRouteRowV4(upd, ins *sql.Stmt, route *LiveRoute, pathJSON string) (inserted bool, err error) {
|
// distribution exact. The lookup reads one entry of the prefix index.
|
||||||
if route.V4IPStart == nil || route.V4IPEnd == nil {
|
const (
|
||||||
return false, fmt.Errorf("IPv4 route %s missing range values", route.Prefix)
|
prefixHasLiveRouteV4SQL = `SELECT EXISTS (SELECT 1 FROM live_routes_v4 WHERE prefix = ?)`
|
||||||
}
|
prefixHasLiveRouteV6SQL = `SELECT EXISTS (SELECT 1 FROM live_routes_v6 WHERE prefix = ?)`
|
||||||
|
)
|
||||||
|
|
||||||
res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
|
// upsertRouteRowV4 updates an IPv4 live route, inserting it when no row matched.
|
||||||
*route.V4IPStart, *route.V4IPEnd, route.Prefix, route.OriginASN, route.PeerIP)
|
// It reports whether a new row was inserted and whether that row is the first
|
||||||
if err != nil {
|
// live route for its prefix.
|
||||||
return false, err
|
func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
|
||||||
}
|
inserted, newPrefix bool, err error) {
|
||||||
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,
|
res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
|
||||||
route.Prefix, route.OriginASN, route.PeerIP)
|
route.Prefix, route.OriginASN, route.PeerIP)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
return false, false, err
|
||||||
}
|
}
|
||||||
affected, err := res.RowsAffected()
|
affected, err := res.RowsAffected()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
return false, false, err
|
||||||
}
|
}
|
||||||
if affected > 0 {
|
if affected > 0 {
|
||||||
return false, nil
|
return false, false, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
var prefixHadRoute bool
|
||||||
|
if err := has.QueryRow(route.Prefix).Scan(&prefixHadRoute); err != nil {
|
||||||
|
return false, false, err
|
||||||
}
|
}
|
||||||
|
|
||||||
_, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN,
|
_, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN,
|
||||||
route.PeerIP, pathJSON, route.NextHop, route.LastUpdated)
|
route.PeerIP, pathJSON, route.NextHop, route.LastUpdated)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
return false, false, err
|
||||||
}
|
}
|
||||||
|
|
||||||
return true, nil
|
return true, !prefixHadRoute, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// upsertRouteRowV6 updates an IPv6 live route, 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 upsertRouteRowV6(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
|
||||||
|
inserted, newPrefix 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, false, err
|
||||||
|
}
|
||||||
|
affected, err := res.RowsAffected()
|
||||||
|
if err != nil {
|
||||||
|
return false, false, err
|
||||||
|
}
|
||||||
|
if affected > 0 {
|
||||||
|
return false, false, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
var prefixHadRoute bool
|
||||||
|
if err := has.QueryRow(route.Prefix).Scan(&prefixHadRoute); err != nil {
|
||||||
|
return false, false, err
|
||||||
|
}
|
||||||
|
|
||||||
|
_, err = ins.Exec(route.ID.String(), route.Prefix, route.MaskLength, route.OriginASN,
|
||||||
|
route.PeerIP, pathJSON, route.NextHop, route.LastUpdated)
|
||||||
|
if err != nil {
|
||||||
|
return false, false, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return true, !prefixHadRoute, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// UpsertLiveRouteBatch inserts or updates multiple live routes in a single transaction
|
// UpsertLiveRouteBatch inserts or updates multiple live routes in a single transaction
|
||||||
@@ -335,7 +356,21 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
|
|||||||
}
|
}
|
||||||
defer func() { _ = insV6.Close() }()
|
defer func() { _ = insV6.Close() }()
|
||||||
|
|
||||||
|
hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to prepare IPv4 prefix lookup statement: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = hasV4.Close() }()
|
||||||
|
|
||||||
|
hasV6, err := tx.Prepare(prefixHasLiveRouteV6SQL)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to prepare IPv6 prefix lookup statement: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = hasV6.Close() }()
|
||||||
|
|
||||||
var newV4, newV6 int
|
var newV4, newV6 int
|
||||||
|
// Mask lengths of the prefixes that get their first live route in this batch.
|
||||||
|
var newPrefixMaskLengthsV4, newPrefixMaskLengthsV6 []int
|
||||||
for _, route := range routes {
|
for _, route := range routes {
|
||||||
pathJSON, err := json.Marshal(route.ASPath)
|
pathJSON, err := json.Marshal(route.ASPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -343,24 +378,30 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if route.IPVersion == ipVersionV4 {
|
if route.IPVersion == ipVersionV4 {
|
||||||
inserted, err := upsertRouteRowV4(updV4, insV4, route, string(pathJSON))
|
inserted, newPrefix, err := upsertRouteRowV4(updV4, insV4, hasV4, route, string(pathJSON))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
|
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
|
||||||
}
|
}
|
||||||
if inserted {
|
if inserted {
|
||||||
newV4++
|
newV4++
|
||||||
}
|
}
|
||||||
|
if newPrefix {
|
||||||
|
newPrefixMaskLengthsV4 = append(newPrefixMaskLengthsV4, route.MaskLength)
|
||||||
|
}
|
||||||
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
inserted, err := upsertRouteRowV6(updV6, insV6, route, string(pathJSON))
|
inserted, newPrefix, err := upsertRouteRowV6(updV6, insV6, hasV6, route, string(pathJSON))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
|
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
|
||||||
}
|
}
|
||||||
if inserted {
|
if inserted {
|
||||||
newV6++
|
newV6++
|
||||||
}
|
}
|
||||||
|
if newPrefix {
|
||||||
|
newPrefixMaskLengthsV6 = append(newPrefixMaskLengthsV6, route.MaskLength)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if err = tx.Commit(); err != nil {
|
if err = tx.Commit(); err != nil {
|
||||||
@@ -368,6 +409,7 @@ func (d *Database) UpsertLiveRouteBatch(routes []*LiveRoute) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
d.counts.addRoutes(newV4, newV6)
|
d.counts.addRoutes(newV4, newV6)
|
||||||
|
d.counts.addToDistribution(newPrefixMaskLengthsV4, newPrefixMaskLengthsV6, 1)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -418,8 +460,22 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
|
|||||||
}
|
}
|
||||||
defer func() { _ = stmtV6WithoutOrigin.Close() }()
|
defer func() { _ = stmtV6WithoutOrigin.Close() }()
|
||||||
|
|
||||||
|
hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to prepare IPv4 prefix lookup statement: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = hasV4.Close() }()
|
||||||
|
|
||||||
|
hasV6, err := tx.Prepare(prefixHasLiveRouteV6SQL)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to prepare IPv6 prefix lookup statement: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = hasV6.Close() }()
|
||||||
|
|
||||||
// Process deletions
|
// Process deletions
|
||||||
var deletedV4, deletedV6 int64
|
var deletedV4, deletedV6 int64
|
||||||
|
// Mask lengths of the prefixes this batch leaves with no live route.
|
||||||
|
var gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6 []int
|
||||||
for _, del := range deletions {
|
for _, del := range deletions {
|
||||||
var stmt *sql.Stmt
|
var stmt *sql.Stmt
|
||||||
|
|
||||||
@@ -462,6 +518,31 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
|
|||||||
} else {
|
} else {
|
||||||
deletedV6 += affected
|
deletedV6 += affected
|
||||||
}
|
}
|
||||||
|
if affected == 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
// The prefix leaves the distribution when no live route has it any more.
|
||||||
|
has := hasV4
|
||||||
|
if del.IPVersion != ipVersionV4 {
|
||||||
|
has = hasV6
|
||||||
|
}
|
||||||
|
var prefixHasRoute bool
|
||||||
|
if err := has.QueryRow(del.Prefix).Scan(&prefixHasRoute); err != nil {
|
||||||
|
return fmt.Errorf("failed to look up prefix %s: %w", del.Prefix, err)
|
||||||
|
}
|
||||||
|
if prefixHasRoute {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
maskLength, err := prefixMaskLength(del.Prefix)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if del.IPVersion == ipVersionV4 {
|
||||||
|
gonePrefixMaskLengthsV4 = append(gonePrefixMaskLengthsV4, maskLength)
|
||||||
|
} else {
|
||||||
|
gonePrefixMaskLengthsV6 = append(gonePrefixMaskLengthsV6, maskLength)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if err = tx.Commit(); err != nil {
|
if err = tx.Commit(); err != nil {
|
||||||
@@ -469,6 +550,7 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
d.counts.addRoutes(-int(deletedV4), -int(deletedV6))
|
d.counts.addRoutes(-int(deletedV4), -int(deletedV6))
|
||||||
|
d.counts.addToDistribution(gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6, -1)
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
@@ -1035,16 +1117,16 @@ func (d *Database) GetStats() (Stats, error) {
|
|||||||
|
|
||||||
// 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
|
// The row counts (ASNs, prefixes, peerings, peers, live routes) and the prefix
|
||||||
// in-memory counters, seeded at startup and kept current on every write, so a
|
// distribution come from the in-memory counters, seeded at startup and kept
|
||||||
// read runs no COUNT(*) over the tables. The oldest/newest route timestamps are
|
// current on every write. The oldest/newest route timestamps are read from the
|
||||||
// read from the ends of the last_updated index, and the file size from a
|
// ends of the last_updated index, and the file size from a stat(). No part of
|
||||||
// stat(); neither is a table scan. The only remaining query is the prefix
|
// the read scans a table or a whole index.
|
||||||
// distribution.
|
|
||||||
func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) {
|
func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) {
|
||||||
var stats Stats
|
var stats Stats
|
||||||
|
|
||||||
// Row counts from memory, as a single consistent snapshot.
|
// Row counts and prefix distribution from memory, as a single consistent
|
||||||
|
// snapshot.
|
||||||
d.counts.fill(&stats)
|
d.counts.fill(&stats)
|
||||||
|
|
||||||
// Database file size is a cheap stat() on the file.
|
// Database file size is a cheap stat() on the file.
|
||||||
@@ -1068,17 +1150,6 @@ func (d *Database) GetStatsContext(ctx context.Context) (Stats, error) {
|
|||||||
stats.NewestRoute = newest
|
stats.NewestRoute = newest
|
||||||
}
|
}
|
||||||
|
|
||||||
// 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.
|
|
||||||
d.logger.Warn("Failed to get prefix distribution", "error", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return stats, nil
|
return stats, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1150,9 +1221,9 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
|
|||||||
return fmt.Errorf("failed to encode AS path: %w", err)
|
return fmt.Errorf("failed to encode AS path: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
updateSQL, insertSQL := updateLiveRouteV4SQL, insertLiveRouteV4SQL
|
updateSQL, insertSQL, lookupSQL := updateLiveRouteV4SQL, insertLiveRouteV4SQL, prefixHasLiveRouteV4SQL
|
||||||
if route.IPVersion == ipVersionV6 {
|
if route.IPVersion == ipVersionV6 {
|
||||||
updateSQL, insertSQL = updateLiveRouteV6SQL, insertLiveRouteV6SQL
|
updateSQL, insertSQL, lookupSQL = updateLiveRouteV6SQL, insertLiveRouteV6SQL, prefixHasLiveRouteV6SQL
|
||||||
}
|
}
|
||||||
|
|
||||||
// The write lock is held, so no other writer can insert this key between the
|
// The write lock is held, so no other writer can insert this key between the
|
||||||
@@ -1169,11 +1240,17 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
|
|||||||
}
|
}
|
||||||
defer func() { _ = ins.Close() }()
|
defer func() { _ = ins.Close() }()
|
||||||
|
|
||||||
var inserted bool
|
has, err := d.db.Prepare(lookupSQL)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("failed to prepare prefix lookup statement: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = has.Close() }()
|
||||||
|
|
||||||
|
var inserted, newPrefix bool
|
||||||
if route.IPVersion == ipVersionV4 {
|
if route.IPVersion == ipVersionV4 {
|
||||||
inserted, err = upsertRouteRowV4(upd, ins, route, string(pathJSON))
|
inserted, newPrefix, err = upsertRouteRowV4(upd, ins, has, route, string(pathJSON))
|
||||||
} else {
|
} else {
|
||||||
inserted, err = upsertRouteRowV6(upd, ins, route, string(pathJSON))
|
inserted, newPrefix, err = upsertRouteRowV6(upd, ins, has, route, string(pathJSON))
|
||||||
}
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
|
return fmt.Errorf("failed to upsert route %s: %w", route.Prefix, err)
|
||||||
@@ -1186,6 +1263,13 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
|
|||||||
d.counts.addRoutes(0, 1)
|
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
|
return nil
|
||||||
}
|
}
|
||||||
@@ -1204,21 +1288,33 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string)
|
|||||||
|
|
||||||
isV4 := ipnet.IP.To4() != nil
|
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 func() {
|
||||||
|
if err := tx.Rollback(); err != nil && err != sql.ErrTxDone {
|
||||||
|
d.logger.Error("Failed to rollback transaction", "error", err)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
// Literal per-table queries (rather than one formatted with the table name)
|
// 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
|
// so the delete carries no dynamically built SQL. A delete with no origin
|
||||||
// ASN can remove several rows.
|
// ASN can remove several rows.
|
||||||
var res sql.Result
|
var res sql.Result
|
||||||
switch {
|
switch {
|
||||||
case isV4 && originASN == 0:
|
case isV4 && originASN == 0:
|
||||||
res, err = d.db.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
|
res, err = tx.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
|
||||||
case isV4:
|
case isV4:
|
||||||
res, err = d.db.Exec(
|
res, err = tx.Exec(
|
||||||
`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
|
`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
|
||||||
prefix, originASN, peerIP)
|
prefix, originASN, peerIP)
|
||||||
case originASN == 0:
|
case originASN == 0:
|
||||||
res, err = d.db.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
|
res, err = tx.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
|
||||||
default:
|
default:
|
||||||
res, err = d.db.Exec(
|
res, err = tx.Exec(
|
||||||
`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
|
`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
|
||||||
prefix, originASN, peerIP)
|
prefix, originASN, peerIP)
|
||||||
}
|
}
|
||||||
@@ -1230,11 +1326,37 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string)
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
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
|
||||||
|
if err := tx.QueryRow(lookupSQL, prefix).Scan(&prefixHasRoute); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := tx.Commit(); err != nil {
|
||||||
|
return fmt.Errorf("failed to commit transaction: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
if isV4 {
|
if isV4 {
|
||||||
d.counts.addRoutes(-int(affected), 0)
|
d.counts.addRoutes(-int(affected), 0)
|
||||||
} else {
|
} else {
|
||||||
d.counts.addRoutes(0, -int(affected))
|
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
|
return nil
|
||||||
}
|
}
|
||||||
@@ -1244,7 +1366,9 @@ func (d *Database) GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []Pr
|
|||||||
return d.GetPrefixDistributionContext(context.Background())
|
return d.GetPrefixDistributionContext(context.Background())
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetPrefixDistributionContext returns the distribution of unique prefixes by mask length with context support
|
// 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) (
|
func (d *Database) GetPrefixDistributionContext(ctx context.Context) (
|
||||||
ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) {
|
ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error) {
|
||||||
// IPv4 distribution - count unique prefixes from v4 table
|
// IPv4 distribution - count unique prefixes from v4 table
|
||||||
@@ -1271,6 +1395,10 @@ func (d *Database) GetPrefixDistributionContext(ctx context.Context) (
|
|||||||
}
|
}
|
||||||
ipv4 = append(ipv4, dist)
|
ipv4 = append(ipv4, dist)
|
||||||
}
|
}
|
||||||
|
// A read that stops partway ends the loop without an error of its own.
|
||||||
|
if err := rows4.Err(); err != nil {
|
||||||
|
return nil, nil, fmt.Errorf("failed to read IPv4 distribution: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
// IPv6 distribution - count unique prefixes from v6 table
|
// IPv6 distribution - count unique prefixes from v6 table
|
||||||
query = `
|
query = `
|
||||||
@@ -1296,6 +1424,9 @@ func (d *Database) GetPrefixDistributionContext(ctx context.Context) (
|
|||||||
}
|
}
|
||||||
ipv6 = append(ipv6, dist)
|
ipv6 = append(ipv6, dist)
|
||||||
}
|
}
|
||||||
|
if err := rows6.Err(); err != nil {
|
||||||
|
return nil, nil, fmt.Errorf("failed to read IPv6 distribution: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
return ipv4, ipv6, nil
|
return ipv4, ipv6, nil
|
||||||
}
|
}
|
||||||
@@ -1322,176 +1453,6 @@ func (d *Database) GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ip
|
|||||||
return ipv4Count, ipv6Count, nil
|
return ipv4Count, ipv6Count, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetASInfoForIP returns AS information for the given IP address
|
|
||||||
func (d *Database) GetASInfoForIP(ip string) (*ASInfo, error) {
|
|
||||||
return d.GetASInfoForIPContext(context.Background(), ip)
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetASInfoForIPContext returns AS information for the given IP address with context support
|
|
||||||
func (d *Database) GetASInfoForIPContext(ctx context.Context, ip string) (*ASInfo, error) {
|
|
||||||
// Parse the IP to validate it
|
|
||||||
parsedIP := net.ParseIP(ip)
|
|
||||||
if parsedIP == nil {
|
|
||||||
return nil, fmt.Errorf("%w: %s", ErrInvalidIP, ip)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Determine IP version
|
|
||||||
ipVersion := ipVersionV4
|
|
||||||
ipv4 := parsedIP.To4()
|
|
||||||
if ipv4 == nil {
|
|
||||||
ipVersion = ipVersionV6
|
|
||||||
}
|
|
||||||
|
|
||||||
// For IPv4, use optimized range query
|
|
||||||
if ipVersion == ipVersionV4 {
|
|
||||||
// Convert IP to 32-bit unsigned integer
|
|
||||||
ipUint := ipToUint32(ipv4)
|
|
||||||
|
|
||||||
query := `
|
|
||||||
SELECT DISTINCT lr.prefix, lr.mask_length, lr.origin_asn, lr.last_updated, a.handle, a.description
|
|
||||||
FROM live_routes_v4 lr
|
|
||||||
LEFT JOIN asns a ON a.asn = lr.origin_asn
|
|
||||||
WHERE lr.ip_start <= ? AND lr.ip_end >= ?
|
|
||||||
ORDER BY lr.mask_length DESC
|
|
||||||
LIMIT 1
|
|
||||||
`
|
|
||||||
|
|
||||||
var prefix string
|
|
||||||
var maskLength, originASN int
|
|
||||||
var lastUpdated time.Time
|
|
||||||
var handle, description sql.NullString
|
|
||||||
|
|
||||||
err := d.db.QueryRowContext(ctx, query, ipUint, ipUint).Scan(
|
|
||||||
&prefix, &maskLength, &originASN, &lastUpdated, &handle, &description)
|
|
||||||
if err != nil {
|
|
||||||
if err == sql.ErrNoRows {
|
|
||||||
return nil, fmt.Errorf("%w for IP %s", ErrNoRoute, ip)
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil, fmt.Errorf("failed to query routes: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
age := time.Since(lastUpdated).Round(time.Second).String()
|
|
||||||
|
|
||||||
return &ASInfo{
|
|
||||||
ASN: originASN,
|
|
||||||
Handle: handle.String,
|
|
||||||
Description: description.String,
|
|
||||||
Prefix: prefix,
|
|
||||||
LastUpdated: lastUpdated,
|
|
||||||
Age: age,
|
|
||||||
}, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// For IPv6, use the original method since we don't have range optimization
|
|
||||||
query := `
|
|
||||||
SELECT DISTINCT lr.prefix, lr.mask_length, lr.origin_asn, lr.last_updated, a.handle, a.description
|
|
||||||
FROM live_routes_v6 lr
|
|
||||||
LEFT JOIN asns a ON a.asn = lr.origin_asn
|
|
||||||
ORDER BY lr.mask_length DESC
|
|
||||||
`
|
|
||||||
|
|
||||||
rows, err := d.db.QueryContext(ctx, query)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("failed to query routes: %w", err)
|
|
||||||
}
|
|
||||||
defer func() { _ = rows.Close() }()
|
|
||||||
|
|
||||||
// Find the most specific matching prefix
|
|
||||||
var bestMatch struct {
|
|
||||||
prefix string
|
|
||||||
maskLength int
|
|
||||||
originASN int
|
|
||||||
lastUpdated time.Time
|
|
||||||
handle sql.NullString
|
|
||||||
description sql.NullString
|
|
||||||
}
|
|
||||||
bestMaskLength := -1
|
|
||||||
|
|
||||||
for rows.Next() {
|
|
||||||
var prefix string
|
|
||||||
var maskLength, originASN int
|
|
||||||
var lastUpdated time.Time
|
|
||||||
var handle, description sql.NullString
|
|
||||||
|
|
||||||
if err := rows.Scan(&prefix, &maskLength, &originASN, &lastUpdated, &handle, &description); err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
// Parse the prefix CIDR
|
|
||||||
_, ipNet, err := net.ParseCIDR(prefix)
|
|
||||||
if err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check if the IP is in this prefix
|
|
||||||
if ipNet.Contains(parsedIP) && maskLength > bestMaskLength {
|
|
||||||
bestMatch.prefix = prefix
|
|
||||||
bestMatch.maskLength = maskLength
|
|
||||||
bestMatch.originASN = originASN
|
|
||||||
bestMatch.lastUpdated = lastUpdated
|
|
||||||
bestMatch.handle = handle
|
|
||||||
bestMatch.description = description
|
|
||||||
bestMaskLength = maskLength
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if bestMaskLength == -1 {
|
|
||||||
return nil, fmt.Errorf("%w for IP %s", ErrNoRoute, ip)
|
|
||||||
}
|
|
||||||
|
|
||||||
age := time.Since(bestMatch.lastUpdated).Round(time.Second).String()
|
|
||||||
|
|
||||||
return &ASInfo{
|
|
||||||
ASN: bestMatch.originASN,
|
|
||||||
Handle: bestMatch.handle.String,
|
|
||||||
Description: bestMatch.description.String,
|
|
||||||
Prefix: bestMatch.prefix,
|
|
||||||
LastUpdated: bestMatch.lastUpdated,
|
|
||||||
Age: age,
|
|
||||||
}, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// ipToUint32 converts an IPv4 address to a 32-bit unsigned integer
|
|
||||||
func ipToUint32(ip net.IP) uint32 {
|
|
||||||
if len(ip) == ipv6Length {
|
|
||||||
// Convert to 4-byte representation
|
|
||||||
ip = ip[ipv4Offset:ipv6Length]
|
|
||||||
}
|
|
||||||
|
|
||||||
return uint32(ip[0])<<24 | uint32(ip[1])<<16 | uint32(ip[2])<<8 | uint32(ip[3])
|
|
||||||
}
|
|
||||||
|
|
||||||
// CalculateIPv4Range calculates the start and end IP addresses for an IPv4 CIDR block
|
|
||||||
func CalculateIPv4Range(cidr string) (start, end uint32, err error) {
|
|
||||||
_, ipNet, err := net.ParseCIDR(cidr)
|
|
||||||
if err != nil {
|
|
||||||
return 0, 0, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Get the network address (start of range)
|
|
||||||
ip := ipNet.IP.To4()
|
|
||||||
if ip == nil {
|
|
||||||
return 0, 0, fmt.Errorf("not an IPv4 address")
|
|
||||||
}
|
|
||||||
|
|
||||||
start = ipToUint32(ip)
|
|
||||||
|
|
||||||
// Calculate the end of the range
|
|
||||||
ones, bits := ipNet.Mask.Size()
|
|
||||||
hostBits := bits - ones
|
|
||||||
if hostBits >= ipv4Bits {
|
|
||||||
// Special case for /0 - entire IPv4 space
|
|
||||||
end = maxIPv4
|
|
||||||
} else {
|
|
||||||
// Safe to convert since we checked hostBits < 32
|
|
||||||
//nolint:gosec // hostBits is guaranteed to be < 32 from the check above
|
|
||||||
end = start | ((1 << uint(hostBits)) - 1)
|
|
||||||
}
|
|
||||||
|
|
||||||
return start, end, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetASDetails returns detailed information about an AS including prefixes
|
// GetASDetails returns detailed information about an AS including prefixes
|
||||||
func (d *Database) GetASDetails(asn int) (*ASN, []LiveRoute, error) {
|
func (d *Database) GetASDetails(asn int) (*ASN, []LiveRoute, error) {
|
||||||
return d.GetASDetailsContext(context.Background(), asn)
|
return d.GetASDetailsContext(context.Background(), asn)
|
||||||
@@ -1913,39 +1874,37 @@ func (d *Database) GetIPInfo(ip string) (*IPInfo, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// GetIPInfoContext returns comprehensive IP information with context support.
|
// 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 0
|
||||||
|
// it looks up the address's network prefix at that length, and the first length
|
||||||
|
// with a live route wins. That is at most 33 or 129 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) {
|
func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error) {
|
||||||
// Parse the IP to validate it
|
addr, err := netip.ParseAddr(ip)
|
||||||
parsedIP := net.ParseIP(ip)
|
if err != nil {
|
||||||
if parsedIP == nil {
|
|
||||||
return nil, fmt.Errorf("%w: %s", ErrInvalidIP, ip)
|
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()
|
||||||
|
|
||||||
// Determine IP version
|
info := &IPInfo{IP: ip, IPVersion: ipVersionV6}
|
||||||
ipv4 := parsedIP.To4()
|
routesTable, prefixesTable := "live_routes_v6", "prefixes_v6"
|
||||||
if ipv4 != nil {
|
if addr.Is4() {
|
||||||
return d.getIPv4Info(ctx, ip, ipv4)
|
info.IPVersion = ipVersionV4
|
||||||
|
routesTable, prefixesTable = "live_routes_v4", "prefixes_v4"
|
||||||
}
|
}
|
||||||
|
|
||||||
return d.getIPv6Info(ctx, ip, parsedIP)
|
// Route info with peer count and prefix first_seen
|
||||||
}
|
//nolint:gosec // Table names are hardcoded based on IP version
|
||||||
|
query := fmt.Sprintf(`
|
||||||
// getIPv4Info returns comprehensive IP information for an IPv4 address.
|
|
||||||
func (d *Database) getIPv4Info(ctx context.Context, ip string, ipv4 net.IP) (*IPInfo, error) {
|
|
||||||
info := &IPInfo{
|
|
||||||
IP: ip,
|
|
||||||
IPVersion: ipVersionV4,
|
|
||||||
}
|
|
||||||
|
|
||||||
ipUint := ipToUint32(ipv4)
|
|
||||||
|
|
||||||
// Get route info with peer count and prefix first_seen
|
|
||||||
query := `
|
|
||||||
SELECT
|
SELECT
|
||||||
lr.prefix,
|
lr.prefix,
|
||||||
lr.mask_length,
|
lr.mask_length,
|
||||||
lr.origin_asn,
|
lr.origin_asn,
|
||||||
lr.last_updated,
|
lr.last_updated,
|
||||||
(SELECT COUNT(DISTINCT peer_ip) FROM live_routes_v4 WHERE prefix = lr.prefix) as num_peers,
|
(SELECT COUNT(DISTINCT peer_ip) FROM %[1]s WHERE prefix = lr.prefix) as num_peers,
|
||||||
p.first_seen,
|
p.first_seen,
|
||||||
a.handle,
|
a.handle,
|
||||||
a.description,
|
a.description,
|
||||||
@@ -1957,19 +1916,27 @@ func (d *Database) getIPv4Info(ctx context.Context, ip string, ipv4 net.IP) (*IP
|
|||||||
a.abuse_email,
|
a.abuse_email,
|
||||||
a.rir,
|
a.rir,
|
||||||
a.whois_updated_at
|
a.whois_updated_at
|
||||||
FROM live_routes_v4 lr
|
FROM %[1]s lr
|
||||||
LEFT JOIN prefixes_v4 p ON p.prefix = lr.prefix
|
LEFT JOIN %[2]s p ON p.prefix = lr.prefix
|
||||||
LEFT JOIN asns a ON a.asn = lr.origin_asn
|
LEFT JOIN asns a ON a.asn = lr.origin_asn
|
||||||
WHERE lr.ip_start <= ? AND lr.ip_end >= ?
|
WHERE lr.prefix = ?
|
||||||
ORDER BY lr.mask_length DESC
|
|
||||||
LIMIT 1
|
LIMIT 1
|
||||||
`
|
`, 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, address, countryCode, abuseEmail, rir sql.NullString
|
var handle, description, asName, orgName, orgID, address, countryCode, abuseEmail, rir sql.NullString
|
||||||
var prefixFirstSeen sql.NullTime
|
var prefixFirstSeen sql.NullTime
|
||||||
var whoisUpdatedAt sql.NullTime
|
var whoisUpdatedAt sql.NullTime
|
||||||
|
|
||||||
err := d.db.QueryRowContext(ctx, query, ipUint, ipUint).Scan(
|
found := false
|
||||||
|
for bits := addr.BitLen(); bits >= 0; bits-- {
|
||||||
|
prefix := netip.PrefixFrom(addr, bits).Masked().String()
|
||||||
|
err := stmt.QueryRowContext(ctx, prefix).Scan(
|
||||||
&info.Netblock,
|
&info.Netblock,
|
||||||
&info.MaskLength,
|
&info.MaskLength,
|
||||||
&info.ASN,
|
&info.ASN,
|
||||||
@@ -1987,13 +1954,19 @@ func (d *Database) getIPv4Info(ctx context.Context, ip string, ipv4 net.IP) (*IP
|
|||||||
&rir,
|
&rir,
|
||||||
&whoisUpdatedAt,
|
&whoisUpdatedAt,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if errors.Is(err, sql.ErrNoRows) {
|
||||||
if err == sql.ErrNoRows {
|
continue
|
||||||
return nil, fmt.Errorf("%w for IP %s", ErrNoRoute, ip)
|
|
||||||
}
|
}
|
||||||
|
if err != nil {
|
||||||
return nil, fmt.Errorf("failed to query routes: %w", err)
|
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.Handle = handle.String
|
||||||
info.Description = description.String
|
info.Description = description.String
|
||||||
@@ -2016,116 +1989,26 @@ func (d *Database) getIPv4Info(ctx context.Context, ip string, ipv4 net.IP) (*IP
|
|||||||
return info, nil
|
return info, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// getIPv6Info returns comprehensive IP information for an IPv6 address.
|
|
||||||
func (d *Database) getIPv6Info(ctx context.Context, ip string, parsedIP net.IP) (*IPInfo, error) {
|
|
||||||
info := &IPInfo{
|
|
||||||
IP: ip,
|
|
||||||
IPVersion: ipVersionV6,
|
|
||||||
}
|
|
||||||
|
|
||||||
// For IPv6, scan all routes and find best match
|
|
||||||
query := `
|
|
||||||
SELECT DISTINCT
|
|
||||||
lr.prefix,
|
|
||||||
lr.mask_length,
|
|
||||||
lr.origin_asn,
|
|
||||||
lr.last_updated,
|
|
||||||
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 live_routes_v6 lr
|
|
||||||
LEFT JOIN asns a ON a.asn = lr.origin_asn
|
|
||||||
ORDER BY lr.mask_length DESC
|
|
||||||
`
|
|
||||||
|
|
||||||
rows, err := d.db.QueryContext(ctx, query)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("failed to query routes: %w", err)
|
|
||||||
}
|
|
||||||
defer func() { _ = rows.Close() }()
|
|
||||||
|
|
||||||
bestMaskLength := -1
|
|
||||||
|
|
||||||
for rows.Next() {
|
|
||||||
var prefix string
|
|
||||||
var maskLength, originASN int
|
|
||||||
var lastUpdated time.Time
|
|
||||||
var handle, description, asName, orgName, orgID, address, countryCode, abuseEmail, rir sql.NullString
|
|
||||||
var whoisUpdatedAt sql.NullTime
|
|
||||||
|
|
||||||
if err := rows.Scan(
|
|
||||||
&prefix, &maskLength, &originASN, &lastUpdated,
|
|
||||||
&handle, &description, &asName, &orgName, &orgID,
|
|
||||||
&address, &countryCode, &abuseEmail, &rir, &whoisUpdatedAt,
|
|
||||||
); err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
_, ipNet, err := net.ParseCIDR(prefix)
|
|
||||||
if err != nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
if ipNet.Contains(parsedIP) && maskLength > bestMaskLength {
|
|
||||||
info.Netblock = prefix
|
|
||||||
info.MaskLength = maskLength
|
|
||||||
info.ASN = originASN
|
|
||||||
info.LastSeen = lastUpdated
|
|
||||||
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
|
|
||||||
bestMaskLength = maskLength
|
|
||||||
|
|
||||||
if !whoisUpdatedAt.Valid {
|
|
||||||
info.NeedsWHOISRefresh = true
|
|
||||||
} else {
|
|
||||||
const staleThreshold = 30 * 24 * time.Hour
|
|
||||||
info.NeedsWHOISRefresh = time.Since(whoisUpdatedAt.Time) > staleThreshold
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if bestMaskLength == -1 {
|
|
||||||
return nil, fmt.Errorf("%w for IP %s", ErrNoRoute, ip)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Get peer count and first_seen for IPv6
|
|
||||||
countQuery := `SELECT COUNT(DISTINCT peer_ip) FROM live_routes_v6 WHERE prefix = ?`
|
|
||||||
_ = d.db.QueryRowContext(ctx, countQuery, info.Netblock).Scan(&info.NumPeers)
|
|
||||||
|
|
||||||
firstSeenQuery := `SELECT first_seen FROM prefixes_v6 WHERE prefix = ?`
|
|
||||||
var prefixFirstSeen sql.NullTime
|
|
||||||
err = d.db.QueryRowContext(ctx, firstSeenQuery, info.Netblock).Scan(&prefixFirstSeen)
|
|
||||||
if err == nil && prefixFirstSeen.Valid {
|
|
||||||
info.FirstSeen = prefixFirstSeen.Time
|
|
||||||
}
|
|
||||||
|
|
||||||
return info, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Vacuum runs incremental vacuum to reclaim unused pages without blocking writes.
|
// Vacuum runs incremental vacuum to reclaim unused pages without blocking writes.
|
||||||
// It frees up to the specified number of pages per call (0 = all freeable pages).
|
// It frees up to the specified number of pages per call (0 = all freeable pages).
|
||||||
func (d *Database) Vacuum(ctx context.Context) error {
|
func (d *Database) Vacuum(ctx context.Context) error {
|
||||||
// Free up to 1000 pages per call (~4MB with default 4KB page size)
|
// Free up to 1000 pages per call (~4MB with default 4KB page size)
|
||||||
// This keeps each vacuum operation quick and non-blocking
|
// This keeps each vacuum operation quick and non-blocking
|
||||||
const pagesToFree = 1000
|
const pagesToFree = 1000
|
||||||
_, err := d.db.ExecContext(ctx, fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree))
|
rows, err := d.db.QueryContext(ctx, fmt.Sprintf("PRAGMA incremental_vacuum(%d)", pagesToFree))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to run incremental vacuum: %w", err)
|
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() {
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return fmt.Errorf("failed to run incremental vacuum: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
+148
-273
@@ -3,23 +3,36 @@ package database
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"net"
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"net/netip"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/config"
|
"git.eeqj.de/sneak/routewatch/internal/config"
|
||||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||||
|
"github.com/google/uuid"
|
||||||
)
|
)
|
||||||
|
|
||||||
// tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in
|
// tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in
|
||||||
// memory"; the DSN change must leave temp_store below this so they spill to disk.
|
// memory"; the DSN change must leave temp_store below this so they spill to disk.
|
||||||
const tempStoreMemory = 2
|
const tempStoreMemory = 2
|
||||||
|
|
||||||
|
// autoVacuumIncremental is the PRAGMA auto_vacuum value meaning "incremental".
|
||||||
|
const autoVacuumIncremental = 2
|
||||||
|
|
||||||
|
// vacuumTestRoutes is how many routes the vacuum test writes and then deletes,
|
||||||
|
// enough to leave many free pages in the file.
|
||||||
|
const vacuumTestRoutes = 2000
|
||||||
|
|
||||||
// heldConnections is how many pooled connections the pragma test holds open at
|
// heldConnections is how many pooled connections the pragma test holds open at
|
||||||
// once so each is a distinct SQLite connection that parsed the DSN.
|
// once so each is a distinct SQLite connection that parsed the DSN.
|
||||||
const heldConnections = 5
|
const heldConnections = 5
|
||||||
|
|
||||||
|
// testPeerIP is the peer every route in the IP lookup test is learned from.
|
||||||
|
const testPeerIP = "192.0.2.254"
|
||||||
|
|
||||||
// Parameters for the checkpoint-contention regression test.
|
// Parameters for the checkpoint-contention regression test.
|
||||||
const (
|
const (
|
||||||
// contentionIterations is how many batch writes race the checkpoint loop.
|
// contentionIterations is how many batch writes race the checkpoint loop.
|
||||||
@@ -32,286 +45,98 @@ const (
|
|||||||
asnSecondBand = 100
|
asnSecondBand = 100
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestIPToUint32(t *testing.T) {
|
// TestGetIPInfoFindsMostSpecificLiveRoute stores nested live prefixes for both
|
||||||
tests := []struct {
|
// families and checks that a lookup returns the most specific one covering the
|
||||||
name string
|
// address, ErrNoRoute when none covers it, and the next less specific prefix
|
||||||
ip string
|
// once the only route of the most specific one is withdrawn.
|
||||||
expected uint32
|
func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) {
|
||||||
}{
|
cfg := &config.Config{StateDir: t.TempDir()}
|
||||||
{
|
|
||||||
name: "Simple IP",
|
|
||||||
ip: "192.168.1.1",
|
|
||||||
expected: 3232235777, // 192<<24 + 168<<16 + 1<<8 + 1
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Minimum IP",
|
|
||||||
ip: "0.0.0.0",
|
|
||||||
expected: 0,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Maximum IP",
|
|
||||||
ip: "255.255.255.255",
|
|
||||||
expected: 4294967295,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "10.0.0.0",
|
|
||||||
ip: "10.0.0.0",
|
|
||||||
expected: 167772160,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "172.16.0.0",
|
|
||||||
ip: "172.16.0.0",
|
|
||||||
expected: 2886729728,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "8.8.8.8",
|
|
||||||
ip: "8.8.8.8",
|
|
||||||
expected: 134744072,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "1.2.3.4",
|
|
||||||
ip: "1.2.3.4",
|
|
||||||
expected: 16909060,
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, tt := range tests {
|
db, err := New(cfg, logger.New())
|
||||||
t.Run(tt.name, func(t *testing.T) {
|
if err != nil {
|
||||||
ip := net.ParseIP(tt.ip)
|
t.Fatalf("failed to create database: %v", err)
|
||||||
if ip == nil {
|
|
||||||
t.Fatalf("Failed to parse IP: %s", tt.ip)
|
|
||||||
}
|
}
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
result := ipToUint32(ip)
|
// Nested live prefixes, each originated by its own AS.
|
||||||
if result != tt.expected {
|
origins := map[string]int{
|
||||||
t.Errorf("ipToUint32(%s) = %d, want %d", tt.ip, result, tt.expected)
|
"10.0.0.0/8": 64500,
|
||||||
}
|
"10.1.0.0/16": 64501,
|
||||||
|
"10.1.2.0/24": 64502,
|
||||||
// Test with IPv4-mapped IPv6 address
|
"2001:db8::/32": 64500,
|
||||||
ip6 := net.ParseIP(tt.ip).To16()
|
"2001:db8:1::/48": 64501,
|
||||||
if ip6 != nil {
|
"2001:db8:1:2::/64": 64502,
|
||||||
result6 := ipToUint32(ip6)
|
|
||||||
if result6 != tt.expected {
|
|
||||||
t.Errorf("ipToUint32(%s as IPv6) = %d, want %d", tt.ip, result6, tt.expected)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
routes := make([]*LiveRoute, 0, len(origins))
|
||||||
|
for prefix, asn := range origins {
|
||||||
|
routes = append(routes, &LiveRoute{
|
||||||
|
ID: uuid.New(),
|
||||||
|
Prefix: prefix,
|
||||||
|
MaskLength: netip.MustParsePrefix(prefix).Bits(),
|
||||||
|
IPVersion: detectIPVersion(prefix),
|
||||||
|
OriginASN: asn,
|
||||||
|
PeerIP: testPeerIP,
|
||||||
|
ASPath: []int{asn},
|
||||||
|
NextHop: testPeerIP,
|
||||||
|
LastUpdated: ts,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
if err := db.UpsertLiveRouteBatch(routes); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestCalculateIPv4Range(t *testing.T) {
|
// lookup checks that ip resolves to the live prefix want, or to ErrNoRoute
|
||||||
tests := []struct {
|
// when want is empty.
|
||||||
name string
|
lookup := func(ip, want string) {
|
||||||
cidr string
|
t.Helper()
|
||||||
wantStart uint32
|
|
||||||
wantEnd uint32
|
info, err := db.GetIPInfo(ip)
|
||||||
wantErr bool
|
if want == "" {
|
||||||
}{
|
if !errors.Is(err, ErrNoRoute) {
|
||||||
{
|
t.Errorf("GetIPInfo(%s) = %+v, %v; want ErrNoRoute", ip, info, err)
|
||||||
name: "Single IP /32",
|
|
||||||
cidr: "192.168.1.1/32",
|
|
||||||
wantStart: 3232235777,
|
|
||||||
wantEnd: 3232235777,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Class C /24",
|
|
||||||
cidr: "192.168.1.0/24",
|
|
||||||
wantStart: 3232235776, // 192.168.1.0
|
|
||||||
wantEnd: 3232236031, // 192.168.1.255
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Class B /16",
|
|
||||||
cidr: "192.168.0.0/16",
|
|
||||||
wantStart: 3232235520, // 192.168.0.0
|
|
||||||
wantEnd: 3232301055, // 192.168.255.255
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Class A /8",
|
|
||||||
cidr: "10.0.0.0/8",
|
|
||||||
wantStart: 167772160, // 10.0.0.0
|
|
||||||
wantEnd: 184549375, // 10.255.255.255
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Entire IPv4 space /0",
|
|
||||||
cidr: "0.0.0.0/0",
|
|
||||||
wantStart: 0,
|
|
||||||
wantEnd: 4294967295,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Small subnet /30",
|
|
||||||
cidr: "192.168.1.0/30",
|
|
||||||
wantStart: 3232235776, // 192.168.1.0
|
|
||||||
wantEnd: 3232235779, // 192.168.1.3
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Medium subnet /20",
|
|
||||||
cidr: "172.16.0.0/20",
|
|
||||||
wantStart: 2886729728, // 172.16.0.0
|
|
||||||
wantEnd: 2886733823, // 172.16.15.255
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Private range 172.16/12",
|
|
||||||
cidr: "172.16.0.0/12",
|
|
||||||
wantStart: 2886729728, // 172.16.0.0
|
|
||||||
wantEnd: 2887778303, // 172.31.255.255
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Google DNS /29",
|
|
||||||
cidr: "8.8.8.8/29",
|
|
||||||
wantStart: 134744072, // 8.8.8.8 (network is actually 8.8.8.8 with /29)
|
|
||||||
wantEnd: 134744079, // 8.8.8.15
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Non-zero host bits",
|
|
||||||
cidr: "192.168.1.5/24",
|
|
||||||
wantStart: 3232235776, // 192.168.1.0 (network address)
|
|
||||||
wantEnd: 3232236031, // 192.168.1.255
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Invalid CIDR",
|
|
||||||
cidr: "192.168.1.1/33",
|
|
||||||
wantErr: true,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Invalid IP",
|
|
||||||
cidr: "256.256.256.256/24",
|
|
||||||
wantErr: true,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "IPv6 CIDR",
|
|
||||||
cidr: "2001:db8::/32",
|
|
||||||
wantErr: true,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Empty CIDR",
|
|
||||||
cidr: "",
|
|
||||||
wantErr: true,
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "Missing mask",
|
|
||||||
cidr: "192.168.1.1",
|
|
||||||
wantErr: true,
|
|
||||||
},
|
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, tt := range tests {
|
|
||||||
t.Run(tt.name, func(t *testing.T) {
|
|
||||||
start, end, err := CalculateIPv4Range(tt.cidr)
|
|
||||||
|
|
||||||
if tt.wantErr {
|
|
||||||
if err == nil {
|
|
||||||
t.Errorf("CalculateIPv4Range(%s) expected error, got nil", tt.cidr)
|
|
||||||
}
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Errorf("CalculateIPv4Range(%s) unexpected error: %v", tt.cidr, err)
|
t.Errorf("GetIPInfo(%s): %v", ip, err)
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
if info.Netblock != want || info.MaskLength != netip.MustParsePrefix(want).Bits() ||
|
||||||
if start != tt.wantStart {
|
info.ASN != origins[want] {
|
||||||
t.Errorf("CalculateIPv4Range(%s) start = %d, want %d", tt.cidr, start, tt.wantStart)
|
t.Errorf("GetIPInfo(%s) = %s (mask %d) AS%d, want %s AS%d",
|
||||||
|
ip, info.Netblock, info.MaskLength, info.ASN, want, origins[want])
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if end != tt.wantEnd {
|
lookup("10.1.2.3", "10.1.2.0/24")
|
||||||
t.Errorf("CalculateIPv4Range(%s) end = %d, want %d", tt.cidr, end, tt.wantEnd)
|
lookup("::ffff:10.1.2.3", "10.1.2.0/24")
|
||||||
}
|
lookup("10.1.3.4", "10.1.0.0/16")
|
||||||
|
lookup("10.2.0.1", "10.0.0.0/8")
|
||||||
|
lookup("192.0.2.1", "")
|
||||||
|
lookup("2001:db8:1:2::3", "2001:db8:1:2::/64")
|
||||||
|
lookup("2001:db8:1:3::4", "2001:db8:1::/48")
|
||||||
|
lookup("2001:db8:2::1", "2001:db8::/32")
|
||||||
|
lookup("2001:db9::1", "")
|
||||||
|
|
||||||
// Verify that start <= end
|
err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
|
||||||
if start > end {
|
{Prefix: "10.1.2.0/24", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV4},
|
||||||
t.Errorf("CalculateIPv4Range(%s) start (%d) > end (%d)", tt.cidr, start, end)
|
{Prefix: "2001:db8:1:2::/64", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV6},
|
||||||
}
|
|
||||||
|
|
||||||
// Verify the range size matches the CIDR mask
|
|
||||||
if !tt.wantErr && tt.cidr != "" {
|
|
||||||
_, ipNet, _ := net.ParseCIDR(tt.cidr)
|
|
||||||
if ipNet != nil {
|
|
||||||
ones, bits := ipNet.Mask.Size()
|
|
||||||
expectedSize := uint32(1) << uint(bits-ones)
|
|
||||||
actualSize := end - start + 1
|
|
||||||
if actualSize != expectedSize {
|
|
||||||
t.Errorf("CalculateIPv4Range(%s) range size = %d, want %d", tt.cidr, actualSize, expectedSize)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
})
|
})
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestIPv4RangeIntegration(t *testing.T) {
|
|
||||||
// Test that our functions work correctly together
|
|
||||||
tests := []struct {
|
|
||||||
name string
|
|
||||||
cidr string
|
|
||||||
testIPs []string
|
|
||||||
shouldContain []bool
|
|
||||||
}{
|
|
||||||
{
|
|
||||||
name: "192.168.1.0/24",
|
|
||||||
cidr: "192.168.1.0/24",
|
|
||||||
testIPs: []string{
|
|
||||||
"192.168.1.0",
|
|
||||||
"192.168.1.1",
|
|
||||||
"192.168.1.255",
|
|
||||||
"192.168.0.255",
|
|
||||||
"192.168.2.0",
|
|
||||||
},
|
|
||||||
shouldContain: []bool{true, true, true, false, false},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "10.0.0.0/8",
|
|
||||||
cidr: "10.0.0.0/8",
|
|
||||||
testIPs: []string{
|
|
||||||
"10.0.0.0",
|
|
||||||
"10.255.255.255",
|
|
||||||
"10.1.2.3",
|
|
||||||
"9.255.255.255",
|
|
||||||
"11.0.0.0",
|
|
||||||
},
|
|
||||||
shouldContain: []bool{true, true, true, false, false},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
name: "172.16.0.0/12",
|
|
||||||
cidr: "172.16.0.0/12",
|
|
||||||
testIPs: []string{
|
|
||||||
"172.16.0.0",
|
|
||||||
"172.31.255.255",
|
|
||||||
"172.20.1.1",
|
|
||||||
"172.15.255.255",
|
|
||||||
"172.32.0.0",
|
|
||||||
},
|
|
||||||
shouldContain: []bool{true, true, true, false, false},
|
|
||||||
},
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, tt := range tests {
|
|
||||||
t.Run(tt.name, func(t *testing.T) {
|
|
||||||
start, end, err := CalculateIPv4Range(tt.cidr)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("Failed to calculate range for %s: %v", tt.cidr, err)
|
t.Fatalf("DeleteLiveRouteBatch: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
for i, testIP := range tt.testIPs {
|
lookup("10.1.2.3", "10.1.0.0/16")
|
||||||
ip := net.ParseIP(testIP)
|
lookup("2001:db8:1:2::3", "2001:db8:1::/48")
|
||||||
if ip == nil {
|
|
||||||
t.Fatalf("Failed to parse test IP: %s", testIP)
|
|
||||||
}
|
|
||||||
|
|
||||||
ipUint := ipToUint32(ip)
|
|
||||||
contained := ipUint >= start && ipUint <= end
|
|
||||||
|
|
||||||
if contained != tt.shouldContain[i] {
|
|
||||||
t.Errorf("IP %s in range %s: got %v, want %v", testIP, tt.cidr, contained, tt.shouldContain[i])
|
|
||||||
}
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestConnectionPoolPragmas holds several pooled connections open at once and
|
// TestConnectionPoolPragmas holds several pooled connections open at once and
|
||||||
// checks each one carries the per-connection settings from the DSN, plus the
|
// checks each one carries the per-connection settings from the DSN, plus the
|
||||||
// process-wide hard heap limit.
|
// process-wide hard heap limit, and sees the new file with auto_vacuum
|
||||||
|
// incremental.
|
||||||
func TestConnectionPoolPragmas(t *testing.T) {
|
func TestConnectionPoolPragmas(t *testing.T) {
|
||||||
cfg := &config.Config{StateDir: t.TempDir()}
|
cfg := &config.Config{StateDir: t.TempDir()}
|
||||||
|
|
||||||
@@ -372,6 +197,74 @@ func TestConnectionPoolPragmas(t *testing.T) {
|
|||||||
if hardHeapLimit != sqliteHardHeapLimitBytes {
|
if hardHeapLimit != sqliteHardHeapLimitBytes {
|
||||||
t.Errorf("conn %d: hard_heap_limit = %d, want %d", i, hardHeapLimit, sqliteHardHeapLimitBytes)
|
t.Errorf("conn %d: hard_heap_limit = %d, want %d", i, hardHeapLimit, sqliteHardHeapLimitBytes)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var autoVacuum int
|
||||||
|
if err := c.QueryRowContext(ctx, "PRAGMA auto_vacuum").Scan(&autoVacuum); err != nil {
|
||||||
|
t.Fatalf("conn %d: failed to read auto_vacuum: %v", i, err)
|
||||||
|
}
|
||||||
|
if autoVacuum != autoVacuumIncremental {
|
||||||
|
t.Errorf("conn %d: auto_vacuum = %d, want %d (incremental)", i, autoVacuum, autoVacuumIncremental)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestVacuumReturnsFreePages checks that after routes are deleted from a new
|
||||||
|
// database, one Vacuum call returns every page they used. The deletes leave
|
||||||
|
// fewer free pages than the 1000 Vacuum frees per call, so none may remain.
|
||||||
|
// With auto_vacuum off (issue https://git.eeqj.de/sneak/routewatch/issues/43)
|
||||||
|
// the free pages stayed in the file, and with the PRAGMA run by ExecContext
|
||||||
|
// Vacuum freed only one page per call.
|
||||||
|
func TestVacuumReturnsFreePages(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()
|
||||||
|
ts := time.Now().UTC()
|
||||||
|
const asn = 64500
|
||||||
|
|
||||||
|
routes := make([]*LiveRoute, 0, vacuumTestRoutes)
|
||||||
|
deletions := make([]LiveRouteDeletion, 0, vacuumTestRoutes)
|
||||||
|
for i := range vacuumTestRoutes {
|
||||||
|
route := mkV4Route(t, fmt.Sprintf("10.%d.%d.0/24", i/256, i%256), asn, ts)
|
||||||
|
routes = append(routes, route)
|
||||||
|
deletions = append(deletions, LiveRouteDeletion{
|
||||||
|
Prefix: route.Prefix,
|
||||||
|
OriginASN: asn,
|
||||||
|
PeerIP: route.PeerIP,
|
||||||
|
IPVersion: ipVersionV4,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := db.UpsertLiveRouteBatch(routes); err != nil {
|
||||||
|
t.Fatalf("failed to write routes: %v", err)
|
||||||
|
}
|
||||||
|
if err := db.DeleteLiveRouteBatch(deletions); err != nil {
|
||||||
|
t.Fatalf("failed to delete routes: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var before int
|
||||||
|
if err := db.db.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&before); err != nil {
|
||||||
|
t.Fatalf("failed to read freelist_count: %v", err)
|
||||||
|
}
|
||||||
|
if before == 0 {
|
||||||
|
t.Fatalf("no free pages after deleting %d routes", vacuumTestRoutes)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := db.Vacuum(ctx); err != nil {
|
||||||
|
t.Fatalf("Vacuum failed: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var after int
|
||||||
|
if err := db.db.QueryRowContext(ctx, "PRAGMA freelist_count").Scan(&after); err != nil {
|
||||||
|
t.Fatalf("failed to read freelist_count: %v", err)
|
||||||
|
}
|
||||||
|
if after != 0 {
|
||||||
|
t.Errorf("free pages after Vacuum = %d of %d, want 0", after, before)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -424,21 +317,3 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) {
|
|||||||
cancel()
|
cancel()
|
||||||
wg.Wait()
|
wg.Wait()
|
||||||
}
|
}
|
||||||
|
|
||||||
func BenchmarkIPToUint32(b *testing.B) {
|
|
||||||
ip := net.ParseIP("192.168.1.1")
|
|
||||||
b.ResetTimer()
|
|
||||||
|
|
||||||
for i := 0; i < b.N; i++ {
|
|
||||||
_ = ipToUint32(ip)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func BenchmarkCalculateIPv4Range(b *testing.B) {
|
|
||||||
cidr := "192.168.0.0/16"
|
|
||||||
b.ResetTimer()
|
|
||||||
|
|
||||||
for i := 0; i < b.N; i++ {
|
|
||||||
_, _, _ = CalculateIPv4Range(cidr)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -63,8 +63,6 @@ type Store interface {
|
|||||||
GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error)
|
GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error)
|
||||||
|
|
||||||
// IP lookup operations
|
// IP lookup operations
|
||||||
GetASInfoForIP(ip string) (*ASInfo, error)
|
|
||||||
GetASInfoForIPContext(ctx context.Context, ip string) (*ASInfo, error)
|
|
||||||
GetIPInfo(ip string) (*IPInfo, error)
|
GetIPInfo(ip string) (*IPInfo, error)
|
||||||
GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error)
|
GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error)
|
||||||
|
|
||||||
|
|||||||
@@ -77,9 +77,6 @@ type LiveRoute struct {
|
|||||||
ASPath []int `json:"as_path"`
|
ASPath []int `json:"as_path"`
|
||||||
NextHop string `json:"next_hop"`
|
NextHop string `json:"next_hop"`
|
||||||
LastUpdated time.Time `json:"last_updated"`
|
LastUpdated time.Time `json:"last_updated"`
|
||||||
// IPv4 range fields for fast lookups (nil for IPv6)
|
|
||||||
V4IPStart *uint32 `json:"v4_ip_start,omitempty"`
|
|
||||||
V4IPEnd *uint32 `json:"v4_ip_end,omitempty"`
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// PrefixDistribution represents the distribution of prefixes by mask length
|
// PrefixDistribution represents the distribution of prefixes by mask length
|
||||||
@@ -88,16 +85,6 @@ type PrefixDistribution struct {
|
|||||||
Count int `json:"count"`
|
Count int `json:"count"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// ASInfo represents AS information for an IP lookup (legacy format)
|
|
||||||
type ASInfo struct {
|
|
||||||
ASN int `json:"asn"`
|
|
||||||
Handle string `json:"handle"`
|
|
||||||
Description string `json:"description"`
|
|
||||||
Prefix string `json:"prefix"`
|
|
||||||
LastUpdated time.Time `json:"last_updated"`
|
|
||||||
Age string `json:"age"`
|
|
||||||
}
|
|
||||||
|
|
||||||
// IPInfo represents comprehensive IP information for the /ip endpoint
|
// IPInfo represents comprehensive IP information for the /ip endpoint
|
||||||
type IPInfo struct {
|
type IPInfo struct {
|
||||||
IP string `json:"ip"`
|
IP string `json:"ip"`
|
||||||
|
|||||||
@@ -107,9 +107,6 @@ CREATE TABLE IF NOT EXISTS live_routes_v4 (
|
|||||||
as_path TEXT NOT NULL, -- JSON array
|
as_path TEXT NOT NULL, -- JSON array
|
||||||
next_hop TEXT NOT NULL,
|
next_hop TEXT NOT NULL,
|
||||||
last_updated DATETIME NOT NULL,
|
last_updated DATETIME NOT NULL,
|
||||||
-- IPv4 range columns for fast lookups
|
|
||||||
ip_start INTEGER NOT NULL, -- Start of IPv4 range as 32-bit unsigned int
|
|
||||||
ip_end INTEGER NOT NULL, -- End of IPv4 range as 32-bit unsigned int
|
|
||||||
UNIQUE(prefix, origin_asn, peer_ip)
|
UNIQUE(prefix, origin_asn, peer_ip)
|
||||||
);
|
);
|
||||||
|
|
||||||
@@ -123,7 +120,6 @@ CREATE TABLE IF NOT EXISTS live_routes_v6 (
|
|||||||
as_path TEXT NOT NULL, -- JSON array
|
as_path TEXT NOT NULL, -- JSON array
|
||||||
next_hop TEXT NOT NULL,
|
next_hop TEXT NOT NULL,
|
||||||
last_updated DATETIME NOT NULL,
|
last_updated DATETIME NOT NULL,
|
||||||
-- Note: IPv6 doesn't use integer range columns
|
|
||||||
UNIQUE(prefix, origin_asn, peer_ip)
|
UNIQUE(prefix, origin_asn, peer_ip)
|
||||||
);
|
);
|
||||||
|
|
||||||
@@ -132,8 +128,6 @@ CREATE INDEX IF NOT EXISTS idx_live_routes_v4_prefix ON live_routes_v4(prefix);
|
|||||||
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_length ON live_routes_v4(mask_length);
|
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_length ON live_routes_v4(mask_length);
|
||||||
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_asn ON live_routes_v4(origin_asn);
|
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_asn ON live_routes_v4(origin_asn);
|
||||||
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated);
|
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated);
|
||||||
-- Indexes for IPv4 range queries
|
|
||||||
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_ip_range ON live_routes_v4(ip_start, ip_end);
|
|
||||||
-- Index to optimize prefix distribution queries
|
-- Index to optimize prefix distribution queries
|
||||||
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_prefix ON live_routes_v4(mask_length, prefix);
|
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_prefix ON live_routes_v4(mask_length, prefix);
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,8 @@
|
|||||||
package database
|
package database
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"fmt"
|
||||||
|
"net"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
@@ -18,3 +20,14 @@ func detectIPVersion(prefix string) int {
|
|||||||
|
|
||||||
return ipVersionV4
|
return ipVersionV4
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// prefixMaskLength returns the mask length of a prefix such as 192.0.2.0/24.
|
||||||
|
func prefixMaskLength(prefix string) (int, error) {
|
||||||
|
_, network, err := net.ParseCIDR(prefix)
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("invalid prefix %s: %w", prefix, err)
|
||||||
|
}
|
||||||
|
maskLength, _ := network.Mask.Size()
|
||||||
|
|
||||||
|
return maskLength, nil
|
||||||
|
}
|
||||||
|
|||||||
@@ -232,25 +232,6 @@ func (m *mockStore) GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, i
|
|||||||
return m.GetLiveRouteCounts()
|
return m.GetLiveRouteCounts()
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetASInfoForIP mock implementation
|
|
||||||
func (m *mockStore) GetASInfoForIP(ip string) (*database.ASInfo, error) {
|
|
||||||
// Simple mock - return a test AS
|
|
||||||
now := time.Now()
|
|
||||||
return &database.ASInfo{
|
|
||||||
ASN: 15169,
|
|
||||||
Handle: "GOOGLE",
|
|
||||||
Description: "Google LLC",
|
|
||||||
Prefix: "8.8.8.0/24",
|
|
||||||
LastUpdated: now.Add(-5 * time.Minute),
|
|
||||||
Age: "5m0s",
|
|
||||||
}, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetASInfoForIPContext mock implementation with context support
|
|
||||||
func (m *mockStore) GetASInfoForIPContext(ctx context.Context, ip string) (*database.ASInfo, error) {
|
|
||||||
return m.GetASInfoForIP(ip)
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetASDetails mock implementation
|
// GetASDetails mock implementation
|
||||||
func (m *mockStore) GetASDetails(asn int) (*database.ASN, []database.LiveRoute, error) {
|
func (m *mockStore) GetASDetails(asn int) (*database.ASN, []database.LiveRoute, error) {
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package routewatch
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"net"
|
"net"
|
||||||
|
"net/netip"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
@@ -108,7 +109,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
|||||||
for _, announcement := range msg.Announcements {
|
for _, announcement := range msg.Announcements {
|
||||||
for _, prefix := range announcement.Prefixes {
|
for _, prefix := range announcement.Prefixes {
|
||||||
h.batch = append(h.batch, prefixUpdate{
|
h.batch = append(h.batch, prefixUpdate{
|
||||||
prefix: prefix,
|
prefix: canonicalPrefix(prefix),
|
||||||
originASN: originASN,
|
originASN: originASN,
|
||||||
peer: msg.Peer,
|
peer: msg.Peer,
|
||||||
messageType: "announcement",
|
messageType: "announcement",
|
||||||
@@ -125,7 +126,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
|||||||
// Process withdrawals
|
// Process withdrawals
|
||||||
for _, prefix := range msg.Withdrawals {
|
for _, prefix := range msg.Withdrawals {
|
||||||
h.batch = append(h.batch, prefixUpdate{
|
h.batch = append(h.batch, prefixUpdate{
|
||||||
prefix: prefix,
|
prefix: canonicalPrefix(prefix),
|
||||||
originASN: originASN, // Use the originASN from path if available
|
originASN: originASN, // Use the originASN from path if available
|
||||||
peer: msg.Peer,
|
peer: msg.Peer,
|
||||||
messageType: "withdrawal",
|
messageType: "withdrawal",
|
||||||
@@ -264,6 +265,21 @@ func (h *PrefixHandler) flushBatchLocked() {
|
|||||||
h.lastFlush = time.Now()
|
h.lastFlush = time.Now()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// canonicalPrefix returns prefix in the text form net/netip prints, the form
|
||||||
|
// the IP lookup builds when it looks a prefix up. The feed sends IPv6
|
||||||
|
// announcements compressed ("2001:db8::/32") but IPv6 withdrawals uncompressed
|
||||||
|
// ("2001:db8:0:0:0:0:0:0/32"); stored as received, a withdrawal would not match
|
||||||
|
// the route its announcement stored. A prefix that does not parse is returned
|
||||||
|
// unchanged, and the batch flush reports it.
|
||||||
|
func canonicalPrefix(prefix string) string {
|
||||||
|
p, err := netip.ParsePrefix(prefix)
|
||||||
|
if err != nil {
|
||||||
|
return prefix
|
||||||
|
}
|
||||||
|
|
||||||
|
return p.Masked().String()
|
||||||
|
}
|
||||||
|
|
||||||
// parseCIDR extracts the mask length and IP version from a prefix string
|
// parseCIDR extracts the mask length and IP version from a prefix string
|
||||||
func parseCIDR(prefix string) (maskLength int, ipVersion int, err error) {
|
func parseCIDR(prefix string) (maskLength int, ipVersion int, err error) {
|
||||||
_, ipNet, err := net.ParseCIDR(prefix)
|
_, ipNet, err := net.ParseCIDR(prefix)
|
||||||
@@ -315,20 +331,6 @@ func (h *PrefixHandler) processAnnouncement(_ *database.Prefix, update prefixUpd
|
|||||||
LastUpdated: update.timestamp,
|
LastUpdated: update.timestamp,
|
||||||
}
|
}
|
||||||
|
|
||||||
// For IPv4, calculate the IP range
|
|
||||||
if ipVersion == ipv4Version {
|
|
||||||
start, end, err := database.CalculateIPv4Range(update.prefix)
|
|
||||||
if err == nil {
|
|
||||||
liveRoute.V4IPStart = &start
|
|
||||||
liveRoute.V4IPEnd = &end
|
|
||||||
} else {
|
|
||||||
h.logger.Error("Failed to calculate IPv4 range",
|
|
||||||
"prefix", update.prefix,
|
|
||||||
"error", err,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := h.db.UpsertLiveRoute(liveRoute); err != nil {
|
if err := h.db.UpsertLiveRoute(liveRoute); err != nil {
|
||||||
h.logger.Error("Failed to upsert live route",
|
h.logger.Error("Failed to upsert live route",
|
||||||
"prefix", update.prefix,
|
"prefix", update.prefix,
|
||||||
@@ -372,20 +374,6 @@ func (h *PrefixHandler) createLiveRoute(update prefixUpdate) *database.LiveRoute
|
|||||||
LastUpdated: update.timestamp,
|
LastUpdated: update.timestamp,
|
||||||
}
|
}
|
||||||
|
|
||||||
// For IPv4, calculate the IP range
|
|
||||||
if ipVersion == ipv4Version {
|
|
||||||
start, end, err := database.CalculateIPv4Range(update.prefix)
|
|
||||||
if err == nil {
|
|
||||||
liveRoute.V4IPStart = &start
|
|
||||||
liveRoute.V4IPEnd = &end
|
|
||||||
} else {
|
|
||||||
h.logger.Error("Failed to calculate IPv4 range",
|
|
||||||
"prefix", update.prefix,
|
|
||||||
"error", err,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return liveRoute
|
return liveRoute
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -425,20 +413,6 @@ func (h *PrefixHandler) processAnnouncementDirect(update prefixUpdate) {
|
|||||||
LastUpdated: update.timestamp,
|
LastUpdated: update.timestamp,
|
||||||
}
|
}
|
||||||
|
|
||||||
// For IPv4, calculate the IP range
|
|
||||||
if ipVersion == ipv4Version {
|
|
||||||
start, end, err := database.CalculateIPv4Range(update.prefix)
|
|
||||||
if err == nil {
|
|
||||||
liveRoute.V4IPStart = &start
|
|
||||||
liveRoute.V4IPEnd = &end
|
|
||||||
} else {
|
|
||||||
h.logger.Error("Failed to calculate IPv4 range",
|
|
||||||
"prefix", update.prefix,
|
|
||||||
"error", err,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if err := h.db.UpsertLiveRoute(liveRoute); err != nil {
|
if err := h.db.UpsertLiveRoute(liveRoute); err != nil {
|
||||||
h.logger.Error("Failed to upsert live route",
|
h.logger.Error("Failed to upsert live route",
|
||||||
"prefix", update.prefix,
|
"prefix", update.prefix,
|
||||||
|
|||||||
@@ -0,0 +1,74 @@
|
|||||||
|
package routewatch
|
||||||
|
|
||||||
|
import (
|
||||||
|
"errors"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/config"
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/database"
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/ristypes"
|
||||||
|
)
|
||||||
|
|
||||||
|
const testPeerIP = "2001:db8:ffff::1"
|
||||||
|
|
||||||
|
// TestPrefixHandlerStoresPrefixesTheIPLookupFinds runs announcements and
|
||||||
|
// withdrawals through the prefix handler into a real database and looks the
|
||||||
|
// addresses up. The prefixes are written the way the feed sends them: IPv6
|
||||||
|
// announcements compressed, IPv6 withdrawals uncompressed. The withdrawal must
|
||||||
|
// remove the route the announcement stored, and the IP lookup must find the
|
||||||
|
// stored prefix.
|
||||||
|
func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
|
||||||
|
db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to create database: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
// Built without NewPrefixHandler's flush timer, so the test flushes itself.
|
||||||
|
h := &PrefixHandler{db: db, logger: logger.New()}
|
||||||
|
flush := func() {
|
||||||
|
h.mu.Lock()
|
||||||
|
defer h.mu.Unlock()
|
||||||
|
h.flushBatchLocked()
|
||||||
|
}
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
h.HandleMessage(&ristypes.RISMessage{
|
||||||
|
Peer: testPeerIP,
|
||||||
|
Path: ristypes.ASPath{testASNA, testASNB},
|
||||||
|
ParsedTimestamp: ts,
|
||||||
|
Announcements: []ristypes.RISAnnouncement{{
|
||||||
|
NextHop: testPeerIP,
|
||||||
|
Prefixes: []string{"2001:db8:1::/48", "192.0.2.0/24"},
|
||||||
|
}},
|
||||||
|
})
|
||||||
|
flush()
|
||||||
|
|
||||||
|
for ip, want := range map[string]string{
|
||||||
|
"2001:db8:1::1": "2001:db8:1::/48",
|
||||||
|
"192.0.2.1": "192.0.2.0/24",
|
||||||
|
} {
|
||||||
|
info, err := db.GetIPInfo(ip)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetIPInfo(%s) after announcement: %v", ip, err)
|
||||||
|
}
|
||||||
|
if info.Netblock != want || info.ASN != testASNB {
|
||||||
|
t.Errorf("GetIPInfo(%s) = %s AS%d, want %s AS%d", ip, info.Netblock, info.ASN, want, testASNB)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
h.HandleMessage(&ristypes.RISMessage{
|
||||||
|
Peer: testPeerIP,
|
||||||
|
ParsedTimestamp: ts.Add(time.Minute),
|
||||||
|
Withdrawals: []string{"2001:db8:1:0:0:0:0:0/48", "192.0.2.0/24"},
|
||||||
|
})
|
||||||
|
flush()
|
||||||
|
|
||||||
|
for _, ip := range []string{"2001:db8:1::1", "192.0.2.1"} {
|
||||||
|
if info, err := db.GetIPInfo(ip); !errors.Is(err, database.ErrNoRoute) {
|
||||||
|
t.Errorf("GetIPInfo(%s) after withdrawal = %+v, %v; want ErrNoRoute", ip, info, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -2,9 +2,11 @@ package server
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"encoding/json"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"runtime"
|
"runtime"
|
||||||
|
"slices"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -13,6 +15,7 @@ import (
|
|||||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||||
"git.eeqj.de/sneak/routewatch/internal/metrics"
|
"git.eeqj.de/sneak/routewatch/internal/metrics"
|
||||||
"git.eeqj.de/sneak/routewatch/internal/streamer"
|
"git.eeqj.de/sneak/routewatch/internal/streamer"
|
||||||
|
"github.com/google/uuid"
|
||||||
)
|
)
|
||||||
|
|
||||||
// blockingStatsDB embeds database.Store (left nil) and overrides only
|
// blockingStatsDB embeds database.Store (left nil) and overrides only
|
||||||
@@ -71,6 +74,80 @@ func TestStatsHandlersDoNotLeakOnTimeout(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestStatsHandlersAnswerFromMemory checks that both stats handlers answer 200
|
||||||
|
// with the live route counts and the prefix distribution while the database is
|
||||||
|
// closed, so that any query would fail: the request path reads them from
|
||||||
|
// memory. The prefix distribution query it used to run read every live route
|
||||||
|
// and, on a large database, took the whole 4-second deadline, so
|
||||||
|
// /api/v1/stats answered 500 (https://git.eeqj.de/sneak/routewatch/issues/30).
|
||||||
|
// The oldest and newest route times still come from one-row lookups at the ends
|
||||||
|
// of an index; with the database closed they are left out of the answer.
|
||||||
|
func TestStatsHandlersAnswerFromMemory(t *testing.T) {
|
||||||
|
db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("database.New: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*database.LiveRoute{
|
||||||
|
{
|
||||||
|
ID: uuid.New(), Prefix: "198.51.100.0/24", MaskLength: 24, IPVersion: 4,
|
||||||
|
OriginASN: 64500, PeerIP: "192.0.2.1", ASPath: []int{64500}, NextHop: "192.0.2.1",
|
||||||
|
LastUpdated: ts,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
ID: uuid.New(), Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6,
|
||||||
|
OriginASN: 64501, PeerIP: "2001:db8::1", ASPath: []int{64501}, NextHop: "2001:db8::1",
|
||||||
|
LastUpdated: ts,
|
||||||
|
},
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
if err := db.Close(); err != nil {
|
||||||
|
t.Fatalf("Close: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
s := New(db, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
|
||||||
|
handlers := map[string]http.HandlerFunc{
|
||||||
|
"status.json": s.handleStatusJSON(),
|
||||||
|
"stats": s.handleStats(),
|
||||||
|
}
|
||||||
|
|
||||||
|
for name, handler := range handlers {
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
handler(rec, httptest.NewRequest(http.MethodGet, "/", nil))
|
||||||
|
if rec.Code != http.StatusOK {
|
||||||
|
t.Errorf("%s: status %d, want %d; body %s", name, rec.Code, http.StatusOK, rec.Body)
|
||||||
|
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
var body struct {
|
||||||
|
Data struct {
|
||||||
|
IPv4Routes int `json:"ipv4_routes"`
|
||||||
|
IPv6Routes int `json:"ipv6_routes"`
|
||||||
|
IPv4PrefixDistribution []database.PrefixDistribution `json:"ipv4_prefix_distribution"`
|
||||||
|
IPv6PrefixDistribution []database.PrefixDistribution `json:"ipv6_prefix_distribution"`
|
||||||
|
} `json:"data"`
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
|
||||||
|
t.Fatalf("%s: decoding the answer: %v", name, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if body.Data.IPv4Routes != 1 || body.Data.IPv6Routes != 1 {
|
||||||
|
t.Errorf("%s: routes = (v4 %d, v6 %d), want (1, 1)", name, body.Data.IPv4Routes, body.Data.IPv6Routes)
|
||||||
|
}
|
||||||
|
wantV4 := []database.PrefixDistribution{{MaskLength: 24, Count: 1}}
|
||||||
|
if !slices.Equal(body.Data.IPv4PrefixDistribution, wantV4) {
|
||||||
|
t.Errorf("%s: IPv4 distribution = %v, want %v", name, body.Data.IPv4PrefixDistribution, wantV4)
|
||||||
|
}
|
||||||
|
wantV6 := []database.PrefixDistribution{{MaskLength: 32, Count: 1}}
|
||||||
|
if !slices.Equal(body.Data.IPv6PrefixDistribution, wantV6) {
|
||||||
|
t.Errorf("%s: IPv6 distribution = %v, want %v", name, body.Data.IPv6PrefixDistribution, wantV6)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// settledGoroutineCount lets transient goroutines finish, then reports the
|
// settledGoroutineCount lets transient goroutines finish, then reports the
|
||||||
// current count.
|
// current count.
|
||||||
func settledGoroutineCount() int {
|
func settledGoroutineCount() int {
|
||||||
|
|||||||
Reference in New Issue
Block a user