Compare commits
7
Commits
main
..
74bfe17d84
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
74bfe17d84 | ||
|
|
ad4c3d91f1 | ||
|
|
b4dee48401 | ||
|
|
616f13071f | ||
|
|
521c632087 | ||
|
|
77a708d819 | ||
|
|
a43935308e |
@@ -144,8 +144,9 @@ routewatch/
|
|||||||
- **Reconnection**: Exponential backoff (5s-320s) with reset after 30s of stable connection
|
- **Reconnection**: Exponential backoff (5s-320s) with reset after 30s of stable connection
|
||||||
- **IP Lookup**: the most specific live route is found by looking up the
|
- **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
|
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
|
(at most 25 lookups for IPv4, 113 for IPv6); prefixes are stored in the
|
||||||
text form Go's `net/netip` prints
|
text form Go's `net/netip` prints. Routes shorter than /8 for IPv4 or /16
|
||||||
|
for IPv6, such as a default route, are not used.
|
||||||
|
|
||||||
### Database Schema
|
### Database Schema
|
||||||
|
|
||||||
|
|||||||
+209
-42
@@ -8,6 +8,7 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"math/rand/v2"
|
||||||
"net"
|
"net"
|
||||||
"net/netip"
|
"net/netip"
|
||||||
"os"
|
"os"
|
||||||
@@ -37,6 +38,14 @@ const (
|
|||||||
ipv6Bits = 128
|
ipv6Bits = 128
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// The IP lookup does not use routes shorter than these mask lengths. A default
|
||||||
|
// route or a route this short only says where a peer sends traffic it has no
|
||||||
|
// better route for; it does not say who holds the address.
|
||||||
|
const (
|
||||||
|
minLookupMaskLengthV4 = 8
|
||||||
|
minLookupMaskLengthV6 = 16
|
||||||
|
)
|
||||||
|
|
||||||
// SQLite memory tuning. 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
|
||||||
// pooled connection gets them; the heap limits are process-wide and set once.
|
// pooled connection gets them; the heap limits are process-wide and set once.
|
||||||
const (
|
const (
|
||||||
@@ -53,6 +62,17 @@ const (
|
|||||||
sqliteHardHeapLimitBytes = 1610612736
|
sqliteHardHeapLimitBytes = 1610612736
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Bounds on the work GetRandomPrefixesByLength does to pick its sample.
|
||||||
|
const (
|
||||||
|
// maxPrefixesToShuffle is the most prefixes of one mask length that are
|
||||||
|
// read in full and shuffled to pick the sample. A mask length with more is
|
||||||
|
// sampled with random lookups.
|
||||||
|
maxPrefixesToShuffle = 5000
|
||||||
|
// sampleLookupsPerPrefix is the most random lookups made for each prefix
|
||||||
|
// asked for.
|
||||||
|
sampleLookupsPerPrefix = 4
|
||||||
|
)
|
||||||
|
|
||||||
// Common errors
|
// Common errors
|
||||||
var (
|
var (
|
||||||
// ErrInvalidIP is returned when an IP address is malformed
|
// ErrInvalidIP is returned when an IP address is malformed
|
||||||
@@ -1624,16 +1644,17 @@ func (d *Database) GetASPeersContext(ctx context.Context, asn int) ([]ASPeer, er
|
|||||||
}
|
}
|
||||||
|
|
||||||
// GetPrefixDetails returns detailed information about a prefix
|
// GetPrefixDetails returns detailed information about a prefix
|
||||||
func (d *Database) GetPrefixDetails(prefix string) ([]LiveRoute, error) {
|
func (d *Database) GetPrefixDetails(prefix string) ([]LiveRoute, map[int]ASN, error) {
|
||||||
return d.GetPrefixDetailsContext(context.Background(), prefix)
|
return d.GetPrefixDetailsContext(context.Background(), prefix)
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetPrefixDetailsContext returns detailed information about a prefix with context support
|
// GetPrefixDetailsContext returns the live routes for a prefix and, keyed by
|
||||||
func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error) {
|
// AS number, the handle and description of each AS that originates it.
|
||||||
|
func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, map[int]ASN, error) {
|
||||||
// Determine if it's IPv4 or IPv6 by parsing the prefix
|
// Determine if it's IPv4 or IPv6 by parsing the prefix
|
||||||
_, ipnet, err := net.ParseCIDR(prefix)
|
_, ipnet, err := net.ParseCIDR(prefix)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("invalid prefix format: %w", err)
|
return nil, nil, fmt.Errorf("invalid prefix format: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
tableName := "live_routes_v4"
|
tableName := "live_routes_v4"
|
||||||
@@ -1655,11 +1676,12 @@ func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) (
|
|||||||
|
|
||||||
rows, err := d.db.QueryContext(ctx, query, prefix)
|
rows, err := d.db.QueryContext(ctx, query, prefix)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("failed to query prefix details: %w", err)
|
return nil, nil, fmt.Errorf("failed to query prefix details: %w", err)
|
||||||
}
|
}
|
||||||
defer func() { _ = rows.Close() }()
|
defer func() { _ = rows.Close() }()
|
||||||
|
|
||||||
var routes []LiveRoute
|
var routes []LiveRoute
|
||||||
|
origins := make(map[int]ASN)
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
var route LiveRoute
|
var route LiveRoute
|
||||||
var pathJSON string
|
var pathJSON string
|
||||||
@@ -1682,13 +1704,18 @@ func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) (
|
|||||||
route.IPVersion = ipVersion
|
route.IPVersion = ipVersion
|
||||||
route.MaskLength, _ = ipnet.Mask.Size()
|
route.MaskLength, _ = ipnet.Mask.Size()
|
||||||
routes = append(routes, route)
|
routes = append(routes, route)
|
||||||
|
origins[route.OriginASN] = ASN{
|
||||||
|
ASN: route.OriginASN,
|
||||||
|
Handle: handle.String,
|
||||||
|
Description: description.String,
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(routes) == 0 {
|
if len(routes) == 0 {
|
||||||
return nil, fmt.Errorf("%w: %s", ErrNoRoute, prefix)
|
return nil, nil, fmt.Errorf("%w: %s", ErrNoRoute, prefix)
|
||||||
}
|
}
|
||||||
|
|
||||||
return routes, nil
|
return routes, origins, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetRandomPrefixesByLength returns a random sample of prefixes with the specified mask length
|
// GetRandomPrefixesByLength returns a random sample of prefixes with the specified mask length
|
||||||
@@ -1696,45 +1723,183 @@ func (d *Database) GetRandomPrefixesByLength(maskLength, ipVersion, limit int) (
|
|||||||
return d.GetRandomPrefixesByLengthContext(context.Background(), maskLength, ipVersion, limit)
|
return d.GetRandomPrefixesByLengthContext(context.Background(), maskLength, ipVersion, limit)
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetRandomPrefixesByLengthContext returns a random sample of prefixes with context support
|
// GetRandomPrefixesByLengthContext returns limit prefixes with the given mask
|
||||||
|
// length picked at random, or all of them when there are fewer, each once with
|
||||||
|
// its most recently updated live route, for a limit up to
|
||||||
|
// maxPrefixesToShuffle. However many routes there are, it reads at most
|
||||||
|
// maxPrefixesToShuffle+1 prefixes and makes at most sampleLookupsPerPrefix
|
||||||
|
// lookups for each prefix asked for.
|
||||||
func (d *Database) GetRandomPrefixesByLengthContext(
|
func (d *Database) GetRandomPrefixesByLengthContext(
|
||||||
ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error) {
|
ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error) {
|
||||||
// Select unique prefixes with their most recent route information
|
|
||||||
tableName := "live_routes_v4"
|
tableName := "live_routes_v4"
|
||||||
if ipVersion == ipVersionV6 {
|
if ipVersion == ipVersionV6 {
|
||||||
tableName = "live_routes_v6"
|
tableName = "live_routes_v6"
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Reading one prefix more than maxPrefixesToShuffle tells whether these
|
||||||
|
// are all the prefixes with this mask length.
|
||||||
|
prefixes, err := d.firstPrefixesByLength(ctx, tableName, maskLength, maxPrefixesToShuffle+1)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
rand.Shuffle(len(prefixes), func(i, j int) {
|
||||||
|
prefixes[i], prefixes[j] = prefixes[j], prefixes[i]
|
||||||
|
})
|
||||||
|
if len(prefixes) > maxPrefixesToShuffle {
|
||||||
|
prefixes, err = d.samplePrefixesByLength(ctx, tableName, maskLength, limit, prefixes)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
prefixes = prefixes[:min(limit, len(prefixes))]
|
||||||
|
|
||||||
|
return d.newestRouteOfEachPrefix(ctx, tableName, ipVersion, prefixes)
|
||||||
|
}
|
||||||
|
|
||||||
|
// firstPrefixesByLength returns the first n distinct prefixes, in text order,
|
||||||
|
// with the given mask length. It reads only their entries in the mask length
|
||||||
|
// and prefix index.
|
||||||
|
func (d *Database) firstPrefixesByLength(
|
||||||
|
ctx context.Context, tableName string, maskLength, n int) ([]string, error) {
|
||||||
//nolint:gosec // Table name is hardcoded based on IP version
|
//nolint:gosec // Table name is hardcoded based on IP version
|
||||||
query := fmt.Sprintf(`
|
query := fmt.Sprintf(`
|
||||||
WITH unique_prefixes AS (
|
SELECT DISTINCT prefix FROM %s
|
||||||
SELECT prefix, MAX(last_updated) as max_updated
|
WHERE mask_length = ?
|
||||||
FROM %s
|
ORDER BY prefix
|
||||||
WHERE mask_length = ?
|
LIMIT ?
|
||||||
GROUP BY prefix
|
`, tableName)
|
||||||
ORDER BY RANDOM()
|
|
||||||
LIMIT ?
|
|
||||||
)
|
|
||||||
SELECT lr.prefix, lr.mask_length, lr.origin_asn, lr.as_path,
|
|
||||||
lr.peer_ip, lr.last_updated
|
|
||||||
FROM %s lr
|
|
||||||
INNER JOIN unique_prefixes up ON lr.prefix = up.prefix AND lr.last_updated = up.max_updated
|
|
||||||
WHERE lr.mask_length = ?
|
|
||||||
`, tableName, tableName)
|
|
||||||
|
|
||||||
rows, err := d.db.QueryContext(ctx, query, maskLength, limit, maskLength)
|
rows, err := d.db.QueryContext(ctx, query, maskLength, n)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("failed to query random prefixes: %w", err)
|
return nil, fmt.Errorf("failed to list prefixes: %w", err)
|
||||||
}
|
}
|
||||||
defer func() {
|
defer func() { _ = rows.Close() }()
|
||||||
_ = rows.Close()
|
|
||||||
}()
|
|
||||||
|
|
||||||
var routes []LiveRoute
|
var prefixes []string
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
var route LiveRoute
|
var prefix string
|
||||||
|
if err := rows.Scan(&prefix); err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to scan prefix: %w", err)
|
||||||
|
}
|
||||||
|
prefixes = append(prefixes, prefix)
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to list prefixes: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return prefixes, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// samplePrefixesByLength picks limit distinct prefixes with the given mask
|
||||||
|
// length at random, in at most sampleLookupsPerPrefix lookups for each prefix
|
||||||
|
// asked for. ORDER BY RANDOM() would read every route with that mask length.
|
||||||
|
// Instead each lookup draws a row id between those of the first and last route
|
||||||
|
// with that mask length and takes the prefix of the first such route at or
|
||||||
|
// after it. SQLite keeps the row id at the end of every index entry, so the
|
||||||
|
// mask length index is ordered by mask length, then row id, and the lookup is
|
||||||
|
// one search of it.
|
||||||
|
//
|
||||||
|
// The sample is uneven, as the page needs a varied sample, not an even one: a
|
||||||
|
// route is picked in proportion to the row ids between it and the previous
|
||||||
|
// route with that mask length, including the row ids of routes with other mask
|
||||||
|
// lengths and of withdrawn routes. A route that follows a long run of
|
||||||
|
// withdrawn routes is picked far more often than one that directly follows
|
||||||
|
// another route with that mask length.
|
||||||
|
//
|
||||||
|
// Should the lookups keep landing on prefixes already picked, the rest are
|
||||||
|
// taken from shuffled, prefixes of that length in random order.
|
||||||
|
func (d *Database) samplePrefixesByLength(
|
||||||
|
ctx context.Context, tableName string, maskLength, limit int, shuffled []string) ([]string, error) {
|
||||||
|
//nolint:gosec // Table name is hardcoded based on IP version
|
||||||
|
rangeQuery := fmt.Sprintf(`
|
||||||
|
SELECT
|
||||||
|
(SELECT MIN(rowid) FROM %[1]s WHERE mask_length = ?),
|
||||||
|
(SELECT MAX(rowid) FROM %[1]s WHERE mask_length = ?)
|
||||||
|
`, tableName)
|
||||||
|
|
||||||
|
var first, last sql.NullInt64
|
||||||
|
if err := d.db.QueryRowContext(ctx, rangeQuery, maskLength, maskLength).Scan(&first, &last); err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to read row id range: %w", err)
|
||||||
|
}
|
||||||
|
if !first.Valid {
|
||||||
|
// Every route with this mask length was withdrawn since they were listed.
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
//nolint:gosec // Table name is hardcoded based on IP version
|
||||||
|
lookupQuery := fmt.Sprintf(`
|
||||||
|
SELECT prefix FROM %s
|
||||||
|
WHERE mask_length = ? AND rowid >= ?
|
||||||
|
ORDER BY rowid
|
||||||
|
LIMIT 1
|
||||||
|
`, tableName)
|
||||||
|
|
||||||
|
stmt, err := d.db.PrepareContext(ctx, lookupQuery)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to prepare prefix lookup: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = stmt.Close() }()
|
||||||
|
|
||||||
|
picked := make(map[string]bool, limit)
|
||||||
|
prefixes := make([]string, 0, limit)
|
||||||
|
pick := func(prefix string) {
|
||||||
|
if !picked[prefix] {
|
||||||
|
picked[prefix] = true
|
||||||
|
prefixes = append(prefixes, prefix)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for lookups := 0; lookups < limit*sampleLookupsPerPrefix && len(prefixes) < limit; lookups++ {
|
||||||
|
//nolint:gosec // The sample needs variety, not unpredictability.
|
||||||
|
rowID := first.Int64 + rand.Int64N(last.Int64-first.Int64+1)
|
||||||
|
|
||||||
|
var prefix string
|
||||||
|
err := stmt.QueryRowContext(ctx, maskLength, rowID).Scan(&prefix)
|
||||||
|
if errors.Is(err, sql.ErrNoRows) {
|
||||||
|
// The routes from rowID on were withdrawn since the range was read.
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to look up prefix: %w", err)
|
||||||
|
}
|
||||||
|
pick(prefix)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, prefix := range shuffled {
|
||||||
|
if len(prefixes) == limit {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
pick(prefix)
|
||||||
|
}
|
||||||
|
|
||||||
|
return prefixes, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// newestRouteOfEachPrefix returns the most recently updated live route of each
|
||||||
|
// prefix, one lookup on the prefix index per prefix. A prefix whose routes were
|
||||||
|
// all withdrawn since it was picked is left out.
|
||||||
|
func (d *Database) newestRouteOfEachPrefix(
|
||||||
|
ctx context.Context, tableName string, ipVersion int, prefixes []string) ([]LiveRoute, error) {
|
||||||
|
//nolint:gosec // Table name is hardcoded based on IP version
|
||||||
|
query := fmt.Sprintf(`
|
||||||
|
SELECT prefix, mask_length, origin_asn, as_path, peer_ip, last_updated
|
||||||
|
FROM %s
|
||||||
|
WHERE prefix = ?
|
||||||
|
ORDER BY last_updated DESC
|
||||||
|
LIMIT 1
|
||||||
|
`, tableName)
|
||||||
|
|
||||||
|
stmt, err := d.db.PrepareContext(ctx, query)
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to prepare route query: %w", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = stmt.Close() }()
|
||||||
|
|
||||||
|
routes := make([]LiveRoute, 0, len(prefixes))
|
||||||
|
for _, prefix := range prefixes {
|
||||||
|
route := LiveRoute{IPVersion: ipVersion}
|
||||||
var pathJSON string
|
var pathJSON string
|
||||||
err := rows.Scan(
|
err := stmt.QueryRowContext(ctx, prefix).Scan(
|
||||||
&route.Prefix,
|
&route.Prefix,
|
||||||
&route.MaskLength,
|
&route.MaskLength,
|
||||||
&route.OriginASN,
|
&route.OriginASN,
|
||||||
@@ -1742,17 +1907,16 @@ func (d *Database) GetRandomPrefixesByLengthContext(
|
|||||||
&route.PeerIP,
|
&route.PeerIP,
|
||||||
&route.LastUpdated,
|
&route.LastUpdated,
|
||||||
)
|
)
|
||||||
// Set IP version based on which table we queried
|
if errors.Is(err, sql.ErrNoRows) {
|
||||||
route.IPVersion = ipVersion
|
|
||||||
if err != nil {
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("failed to query route of %s: %w", prefix, err)
|
||||||
|
}
|
||||||
|
|
||||||
// Decode AS path
|
|
||||||
if err := json.Unmarshal([]byte(pathJSON), &route.ASPath); err != nil {
|
if err := json.Unmarshal([]byte(pathJSON), &route.ASPath); err != nil {
|
||||||
route.ASPath = []int{}
|
route.ASPath = []int{}
|
||||||
}
|
}
|
||||||
|
|
||||||
routes = append(routes, route)
|
routes = append(routes, route)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1876,11 +2040,12 @@ 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
|
// 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
|
// the prefix index: for each mask length from the longest (32 or 128) down to
|
||||||
// it looks up the address's network prefix at that length, and the first length
|
// minLookupMaskLengthV4 or minLookupMaskLengthV6 it looks up the address's
|
||||||
// with a live route wins. That is at most 33 or 129 indexed lookups, however
|
// network prefix at that length, and the first length with a live route wins.
|
||||||
// many routes are live. The prefix is looked up in the text form net/netip
|
// That is at most 25 or 113 indexed lookups, however many routes are live. The
|
||||||
// prints, which is the form the prefix handler stores.
|
// 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) {
|
||||||
addr, err := netip.ParseAddr(ip)
|
addr, err := netip.ParseAddr(ip)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -1891,9 +2056,11 @@ func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, er
|
|||||||
|
|
||||||
info := &IPInfo{IP: ip, IPVersion: ipVersionV6}
|
info := &IPInfo{IP: ip, IPVersion: ipVersionV6}
|
||||||
routesTable, prefixesTable := "live_routes_v6", "prefixes_v6"
|
routesTable, prefixesTable := "live_routes_v6", "prefixes_v6"
|
||||||
|
minMaskLength := minLookupMaskLengthV6
|
||||||
if addr.Is4() {
|
if addr.Is4() {
|
||||||
info.IPVersion = ipVersionV4
|
info.IPVersion = ipVersionV4
|
||||||
routesTable, prefixesTable = "live_routes_v4", "prefixes_v4"
|
routesTable, prefixesTable = "live_routes_v4", "prefixes_v4"
|
||||||
|
minMaskLength = minLookupMaskLengthV4
|
||||||
}
|
}
|
||||||
|
|
||||||
// Route info with peer count and prefix first_seen
|
// Route info with peer count and prefix first_seen
|
||||||
@@ -1934,7 +2101,7 @@ func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, er
|
|||||||
var whoisUpdatedAt sql.NullTime
|
var whoisUpdatedAt sql.NullTime
|
||||||
|
|
||||||
found := false
|
found := false
|
||||||
for bits := addr.BitLen(); bits >= 0; bits-- {
|
for bits := addr.BitLen(); bits >= minMaskLength; bits-- {
|
||||||
prefix := netip.PrefixFrom(addr, bits).Masked().String()
|
prefix := netip.PrefixFrom(addr, bits).Masked().String()
|
||||||
err := stmt.QueryRowContext(ctx, prefix).Scan(
|
err := stmt.QueryRowContext(ctx, prefix).Scan(
|
||||||
&info.Netblock,
|
&info.Netblock,
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net/netip"
|
"net/netip"
|
||||||
|
"slices"
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -45,6 +46,32 @@ const (
|
|||||||
asnSecondBand = 100
|
asnSecondBand = 100
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Parameters for the prefix-length sample tests.
|
||||||
|
const (
|
||||||
|
// sampleLimit is how many prefixes the tests ask the sample for.
|
||||||
|
sampleLimit = 10
|
||||||
|
// sampledPrefixCount is how many prefixes the sampling test stores, more
|
||||||
|
// than sampleLimit so the sample has to choose among them.
|
||||||
|
sampledPrefixCount = 30
|
||||||
|
// secondV4PeerIP and secondV6PeerIP are the second peer each prefix is
|
||||||
|
// learned from, besides the one mkV4Route and mkV6Route set.
|
||||||
|
secondV4PeerIP = "192.0.2.2"
|
||||||
|
secondV6PeerIP = "2001:db8::2"
|
||||||
|
// largeSamplePrefixCount is how many prefixes the test of a mask length
|
||||||
|
// with more than maxPrefixesToShuffle prefixes stores: twice that, so about
|
||||||
|
// half of them come after the first maxPrefixesToShuffle+1 in text order.
|
||||||
|
largeSamplePrefixCount = 2 * maxPrefixesToShuffle
|
||||||
|
// largeSampleLimit is how many prefixes that test asks for. Each random
|
||||||
|
// lookup lands after the first maxPrefixesToShuffle+1 in text order about
|
||||||
|
// half the time, so none of them doing so is about as likely as
|
||||||
|
// largeSampleLimit coin tosses all coming up heads.
|
||||||
|
largeSampleLimit = 30
|
||||||
|
// liveRoutesV4Table and liveRoutesV6Table are the live route tables the
|
||||||
|
// tests pass to the sampling steps.
|
||||||
|
liveRoutesV4Table = "live_routes_v4"
|
||||||
|
liveRoutesV6Table = "live_routes_v6"
|
||||||
|
)
|
||||||
|
|
||||||
// TestGetIPInfoFindsMostSpecificLiveRoute stores nested live prefixes for both
|
// TestGetIPInfoFindsMostSpecificLiveRoute stores nested live prefixes for both
|
||||||
// families and checks that a lookup returns the most specific one covering the
|
// families and checks that a lookup returns the most specific one covering the
|
||||||
// address, ErrNoRoute when none covers it, and the next less specific prefix
|
// address, ErrNoRoute when none covers it, and the next less specific prefix
|
||||||
@@ -133,6 +160,78 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) {
|
|||||||
lookup("2001:db8:1:2::3", "2001:db8:1::/48")
|
lookup("2001:db8:1:2::3", "2001:db8:1::/48")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestGetIPInfoIgnoresShortRoutes checks that the IP lookup does not report a
|
||||||
|
// default route, or a route shorter than /8 for IPv4 or /16 for IPv6, as the
|
||||||
|
// route for an address, and that a route of exactly /8 or /16, or longer, still
|
||||||
|
// is. The prefixes are written out so that a wrong limit fails the test.
|
||||||
|
func TestGetIPInfoIgnoresShortRoutes(t *testing.T) {
|
||||||
|
cfg := &config.Config{StateDir: t.TempDir()}
|
||||||
|
|
||||||
|
db, err := New(cfg, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to create database: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
announce := func(prefix string, asn int) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
err := db.UpsertLiveRouteBatch([]*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 != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch(%s): %v", prefix, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// lookup checks that ip resolves to the live prefix netblock from asn.
|
||||||
|
lookup := func(ip, netblock string, asn int) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
info, err := db.GetIPInfo(ip)
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("GetIPInfo(%s): %v", ip, err)
|
||||||
|
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if info.Netblock != netblock || info.ASN != asn {
|
||||||
|
t.Errorf("GetIPInfo(%s) = %s AS%d, want %s AS%d", ip, info.Netblock, info.ASN, netblock, asn)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Default and very short routes, and routes one bit shorter than each limit.
|
||||||
|
announce("2000::/3", 64500)
|
||||||
|
announce("0.0.0.0/0", 64500)
|
||||||
|
announce("2000::/15", 64500)
|
||||||
|
announce("192.0.0.0/7", 64500)
|
||||||
|
|
||||||
|
for _, ip := range []string{"2001:db8::1", "192.0.2.1"} {
|
||||||
|
if info, err := db.GetIPInfo(ip); !errors.Is(err, ErrNoRoute) {
|
||||||
|
t.Errorf("GetIPInfo(%s) with only short routes = %+v, %v; want ErrNoRoute", ip, info, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Routes of exactly the minimum length are used.
|
||||||
|
announce("2001::/16", 64501)
|
||||||
|
announce("192.0.0.0/8", 64501)
|
||||||
|
|
||||||
|
lookup("2001:db8::1", "2001::/16", 64501)
|
||||||
|
lookup("192.0.2.1", "192.0.0.0/8", 64501)
|
||||||
|
|
||||||
|
announce("2001:db8::/32", 64502)
|
||||||
|
|
||||||
|
lookup("2001:db8::1", "2001:db8::/32", 64502)
|
||||||
|
}
|
||||||
|
|
||||||
// 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, and sees the new file with auto_vacuum
|
// process-wide hard heap limit, and sees the new file with auto_vacuum
|
||||||
@@ -317,3 +416,281 @@ func TestBatchWriteDuringCheckpoint(t *testing.T) {
|
|||||||
cancel()
|
cancel()
|
||||||
wg.Wait()
|
wg.Wait()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestGetRandomPrefixesByLengthListsPrefixOnce stores one prefix of each
|
||||||
|
// family learned from two peers in the same second and checks that the
|
||||||
|
// prefix-length sample lists each of them once.
|
||||||
|
func TestGetRandomPrefixesByLengthListsPrefixOnce(t *testing.T) {
|
||||||
|
cfg := &config.Config{StateDir: t.TempDir()}
|
||||||
|
|
||||||
|
db, err := New(cfg, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to create database: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
v4First := mkV4Route(t, "198.51.100.0/24", 64500, ts)
|
||||||
|
v4Second := mkV4Route(t, "198.51.100.0/24", 64500, ts)
|
||||||
|
v4Second.PeerIP = secondV4PeerIP
|
||||||
|
v6First := mkV6Route("2001:db8::/32", 64500, ts)
|
||||||
|
v6Second := mkV6Route("2001:db8::/32", 64500, ts)
|
||||||
|
v6Second.PeerIP = secondV6PeerIP
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*LiveRoute{v4First, v4Second, v6First, v6Second}); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, want := range []*LiveRoute{v4First, v6First} {
|
||||||
|
routes, err := db.GetRandomPrefixesByLength(want.MaskLength, want.IPVersion, sampleLimit)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetRandomPrefixesByLength(%d, %d): %v", want.MaskLength, want.IPVersion, err)
|
||||||
|
}
|
||||||
|
if len(routes) != 1 || routes[0].Prefix != want.Prefix {
|
||||||
|
t.Errorf("GetRandomPrefixesByLength(%d, %d) = %+v, want %s listed once",
|
||||||
|
want.MaskLength, want.IPVersion, routes, want.Prefix)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestGetRandomPrefixesByLengthListsNewestRoute stores a prefix learned from
|
||||||
|
// two peers at different times, the older route first, and checks that the
|
||||||
|
// prefix-length sample lists the newer route.
|
||||||
|
func TestGetRandomPrefixesByLengthListsNewestRoute(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() }()
|
||||||
|
|
||||||
|
older := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
olderRoute := mkV4Route(t, "198.51.100.0/24", 64500, older)
|
||||||
|
newerRoute := mkV4Route(t, "198.51.100.0/24", 64500, older.Add(time.Minute))
|
||||||
|
newerRoute.PeerIP = secondV4PeerIP
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*LiveRoute{olderRoute, newerRoute}); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
routes, err := db.GetRandomPrefixesByLength(newerRoute.MaskLength, ipVersionV4, sampleLimit)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetRandomPrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
if len(routes) != 1 || routes[0].PeerIP != newerRoute.PeerIP {
|
||||||
|
t.Errorf("GetRandomPrefixesByLength = %+v, want only the newer route, from %s",
|
||||||
|
routes, newerRoute.PeerIP)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestGetRandomPrefixesByLengthSamplesManyPrefixes stores more /24 prefixes
|
||||||
|
// than the sample asks for, each followed by a /16 so that routes of another
|
||||||
|
// mask length sit between the /24 rows, and each prefix learned from two peers
|
||||||
|
// in the same second. It checks both the sample the pages get, picked from the
|
||||||
|
// shuffled prefixes, and the random lookups used for a mask length with more
|
||||||
|
// than maxPrefixesToShuffle prefixes.
|
||||||
|
func TestGetRandomPrefixesByLengthSamplesManyPrefixes(t *testing.T) {
|
||||||
|
cfg := &config.Config{StateDir: t.TempDir()}
|
||||||
|
|
||||||
|
db, err := New(cfg, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to create database: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
var routes []*LiveRoute
|
||||||
|
var stored []string
|
||||||
|
for i := range sampledPrefixCount {
|
||||||
|
prefix := fmt.Sprintf("10.0.%d.0/24", i)
|
||||||
|
for _, p := range []string{prefix, fmt.Sprintf("172.%d.0.0/16", i)} {
|
||||||
|
first := mkV4Route(t, p, 64500, ts)
|
||||||
|
second := mkV4Route(t, p, 64500, ts)
|
||||||
|
second.PeerIP = secondV4PeerIP
|
||||||
|
routes = append(routes, first, second)
|
||||||
|
}
|
||||||
|
stored = append(stored, prefix)
|
||||||
|
}
|
||||||
|
if err := db.UpsertLiveRouteBatch(routes); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
slices.Sort(stored)
|
||||||
|
|
||||||
|
maskLength := routes[0].MaskLength
|
||||||
|
sample, err := db.GetRandomPrefixesByLength(maskLength, ipVersionV4, sampleLimit)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetRandomPrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
var listed []string
|
||||||
|
for _, route := range sample {
|
||||||
|
listed = append(listed, route.Prefix)
|
||||||
|
}
|
||||||
|
checkPrefixSample(t, "GetRandomPrefixesByLength", listed, stored)
|
||||||
|
|
||||||
|
looked, err := db.samplePrefixesByLength(
|
||||||
|
context.Background(), liveRoutesV4Table, maskLength, sampleLimit, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("samplePrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
checkPrefixSample(t, "samplePrefixesByLength", looked, stored)
|
||||||
|
}
|
||||||
|
|
||||||
|
// checkPrefixSample checks that sample lists sampleLimit prefixes, each once
|
||||||
|
// and each one of stored, and that it is not just the first sampleLimit of
|
||||||
|
// stored, which is in text order.
|
||||||
|
func checkPrefixSample(t *testing.T, name string, sample, stored []string) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
if len(sample) != sampleLimit {
|
||||||
|
t.Errorf("%s listed %d prefixes, want %d", name, len(sample), sampleLimit)
|
||||||
|
}
|
||||||
|
listed := make(map[string]bool)
|
||||||
|
for _, prefix := range sample {
|
||||||
|
if listed[prefix] || !slices.Contains(stored, prefix) {
|
||||||
|
t.Errorf("%s lists %s twice or it is not one of the /24s", name, prefix)
|
||||||
|
}
|
||||||
|
listed[prefix] = true
|
||||||
|
}
|
||||||
|
if slices.Equal(slices.Sorted(slices.Values(sample)), stored[:sampleLimit]) {
|
||||||
|
t.Errorf("%s lists just the first %d prefixes in text order", name, sampleLimit)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestGetRandomPrefixesByLengthSamplesPastFirstPrefixes stores more prefixes
|
||||||
|
// of one mask length than maxPrefixesToShuffle. It checks that
|
||||||
|
// firstPrefixesByLength reads just the first maxPrefixesToShuffle+1 in text
|
||||||
|
// order, and that the sample lists the prefixes asked for, each once, some of
|
||||||
|
// them after those first ones, which only the random lookups reach.
|
||||||
|
func TestGetRandomPrefixesByLengthSamplesPastFirstPrefixes(t *testing.T) {
|
||||||
|
cfg := &config.Config{StateDir: t.TempDir()}
|
||||||
|
|
||||||
|
db, err := New(cfg, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to create database: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
var routes []*LiveRoute
|
||||||
|
var stored []string
|
||||||
|
for i := range largeSamplePrefixCount {
|
||||||
|
route := mkV6Route(fmt.Sprintf("2001:%x::/32", i), 64500, ts)
|
||||||
|
routes = append(routes, route)
|
||||||
|
stored = append(stored, route.Prefix)
|
||||||
|
}
|
||||||
|
if err := db.UpsertLiveRouteBatch(routes); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
slices.Sort(stored)
|
||||||
|
firstPrefixes := stored[:maxPrefixesToShuffle+1]
|
||||||
|
|
||||||
|
maskLength := routes[0].MaskLength
|
||||||
|
read, err := db.firstPrefixesByLength(
|
||||||
|
context.Background(), liveRoutesV6Table, maskLength, maxPrefixesToShuffle+1)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("firstPrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
if !slices.Equal(read, firstPrefixes) {
|
||||||
|
t.Errorf("firstPrefixesByLength read %d prefixes, want the first %d in text order",
|
||||||
|
len(read), len(firstPrefixes))
|
||||||
|
}
|
||||||
|
|
||||||
|
sample, err := db.GetRandomPrefixesByLength(maskLength, ipVersionV6, largeSampleLimit)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetRandomPrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
if len(sample) != largeSampleLimit {
|
||||||
|
t.Errorf("GetRandomPrefixesByLength listed %d prefixes, want %d", len(sample), largeSampleLimit)
|
||||||
|
}
|
||||||
|
listed := make(map[string]bool)
|
||||||
|
pastFirstPrefixes := 0
|
||||||
|
for _, route := range sample {
|
||||||
|
if listed[route.Prefix] {
|
||||||
|
t.Errorf("GetRandomPrefixesByLength lists %s twice", route.Prefix)
|
||||||
|
}
|
||||||
|
listed[route.Prefix] = true
|
||||||
|
if !slices.Contains(firstPrefixes, route.Prefix) {
|
||||||
|
pastFirstPrefixes++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if pastFirstPrefixes == 0 {
|
||||||
|
t.Errorf("GetRandomPrefixesByLength lists only prefixes among the first %d in text order",
|
||||||
|
len(firstPrefixes))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestSamplePrefixesByLengthFillsUpFromShuffled stores a single prefix, so
|
||||||
|
// every random lookup lands on it, and checks that the sample still lists
|
||||||
|
// sampleLimit prefixes: the stored one, then the shuffled ones in order. The
|
||||||
|
// stored prefix is the second shuffled one, so the fill-up must skip it.
|
||||||
|
func TestSamplePrefixesByLengthFillsUpFromShuffled(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() }()
|
||||||
|
|
||||||
|
var shuffled []string
|
||||||
|
for i := range sampledPrefixCount {
|
||||||
|
shuffled = append(shuffled, fmt.Sprintf("10.0.%d.0/24", i))
|
||||||
|
}
|
||||||
|
route := mkV4Route(t, shuffled[1], 64500, time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC))
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*LiveRoute{route}); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
sample, err := db.samplePrefixesByLength(
|
||||||
|
context.Background(), liveRoutesV4Table, route.MaskLength, sampleLimit, shuffled)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("samplePrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
want := append([]string{shuffled[1], shuffled[0]}, shuffled[2:sampleLimit]...)
|
||||||
|
if !slices.Equal(sample, want) {
|
||||||
|
t.Errorf("samplePrefixesByLength = %v, want %v", sample, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestGetRandomPrefixesByLengthSkipsWithdrawnRoutes withdraws a route after
|
||||||
|
// storing it, as can happen between the steps of GetRandomPrefixesByLength. It
|
||||||
|
// checks that the random lookups pick nothing for a mask length with no route
|
||||||
|
// left, and that only the prefixes that still have a route get their newest
|
||||||
|
// route listed.
|
||||||
|
func TestGetRandomPrefixesByLengthSkipsWithdrawnRoutes(t *testing.T) {
|
||||||
|
cfg := &config.Config{StateDir: t.TempDir()}
|
||||||
|
|
||||||
|
db, err := New(cfg, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to create database: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
kept := mkV4Route(t, "198.51.100.0/24", 64500, ts)
|
||||||
|
withdrawn := mkV4Route(t, "172.16.0.0/16", 64500, ts)
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*LiveRoute{kept, withdrawn}); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
if err := db.DeleteLiveRoute(withdrawn.Prefix, withdrawn.OriginASN, withdrawn.PeerIP); err != nil {
|
||||||
|
t.Fatalf("DeleteLiveRoute: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
sample, err := db.samplePrefixesByLength(
|
||||||
|
ctx, liveRoutesV4Table, withdrawn.MaskLength, sampleLimit, []string{withdrawn.Prefix})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("samplePrefixesByLength: %v", err)
|
||||||
|
}
|
||||||
|
if len(sample) != 0 {
|
||||||
|
t.Errorf("samplePrefixesByLength = %v, want none for a mask length with no route left", sample)
|
||||||
|
}
|
||||||
|
|
||||||
|
routes, err := db.newestRouteOfEachPrefix(
|
||||||
|
ctx, liveRoutesV4Table, ipVersionV4, []string{withdrawn.Prefix, kept.Prefix})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("newestRouteOfEachPrefix: %v", err)
|
||||||
|
}
|
||||||
|
if len(routes) != 1 || routes[0].Prefix != kept.Prefix {
|
||||||
|
t.Errorf("newestRouteOfEachPrefix = %+v, want only the route of %s", routes, kept.Prefix)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -76,8 +76,8 @@ type Store interface {
|
|||||||
GetASDetailsContext(ctx context.Context, asn int) (*ASN, []LiveRoute, error)
|
GetASDetailsContext(ctx context.Context, asn int) (*ASN, []LiveRoute, error)
|
||||||
GetASPeers(asn int) ([]ASPeer, error)
|
GetASPeers(asn int) ([]ASPeer, error)
|
||||||
GetASPeersContext(ctx context.Context, asn int) ([]ASPeer, error)
|
GetASPeersContext(ctx context.Context, asn int) ([]ASPeer, error)
|
||||||
GetPrefixDetails(prefix string) ([]LiveRoute, error)
|
GetPrefixDetails(prefix string) ([]LiveRoute, map[int]ASN, error)
|
||||||
GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error)
|
GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, map[int]ASN, error)
|
||||||
GetRandomPrefixesByLength(maskLength, ipVersion, limit int) ([]LiveRoute, error)
|
GetRandomPrefixesByLength(maskLength, ipVersion, limit int) ([]LiveRoute, error)
|
||||||
GetRandomPrefixesByLengthContext(ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error)
|
GetRandomPrefixesByLengthContext(ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error)
|
||||||
|
|
||||||
|
|||||||
+15
-10
@@ -60,17 +60,22 @@ func New() *Tracker {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetConnected updates the connection status
|
// SetConnected updates the connection status. Every connection after the
|
||||||
|
// first counts one reconnect.
|
||||||
func (t *Tracker) SetConnected(connected bool) {
|
func (t *Tracker) SetConnected(connected bool) {
|
||||||
wasConnected := t.isConnected.Swap(connected)
|
t.isConnected.Store(connected)
|
||||||
if connected {
|
if !connected {
|
||||||
t.mu.Lock()
|
return
|
||||||
t.connectedSince = time.Now()
|
}
|
||||||
t.mu.Unlock()
|
|
||||||
// Increment reconnect count (but not for the initial connection)
|
t.mu.Lock()
|
||||||
if wasConnected || t.reconnectCount.Load() > 0 {
|
// connectedSince is zero until the first connection.
|
||||||
t.reconnectCount.Add(1)
|
connectedBefore := !t.connectedSince.IsZero()
|
||||||
}
|
t.connectedSince = time.Now()
|
||||||
|
t.mu.Unlock()
|
||||||
|
|
||||||
|
if connectedBefore {
|
||||||
|
t.reconnectCount.Add(1)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,21 @@
|
|||||||
|
package metrics
|
||||||
|
|
||||||
|
import "testing"
|
||||||
|
|
||||||
|
// TestReconnectCount connects three times with a disconnect between each, the
|
||||||
|
// way the streamer does, and expects every connection after the first to count
|
||||||
|
// one reconnect.
|
||||||
|
func TestReconnectCount(t *testing.T) {
|
||||||
|
tracker := New()
|
||||||
|
|
||||||
|
tracker.SetConnected(true)
|
||||||
|
tracker.SetConnected(false)
|
||||||
|
tracker.SetConnected(true)
|
||||||
|
tracker.SetConnected(false)
|
||||||
|
tracker.SetConnected(true)
|
||||||
|
|
||||||
|
const wantReconnects = 2
|
||||||
|
if got := tracker.GetReconnectCount(); got != wantReconnects {
|
||||||
|
t.Fatalf("reconnect count = %d, want %d", got, wantReconnects)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -252,13 +252,15 @@ func (m *mockStore) GetASDetailsContext(ctx context.Context, asn int) (*database
|
|||||||
}
|
}
|
||||||
|
|
||||||
// GetPrefixDetails mock implementation
|
// GetPrefixDetails mock implementation
|
||||||
func (m *mockStore) GetPrefixDetails(prefix string) ([]database.LiveRoute, error) {
|
func (m *mockStore) GetPrefixDetails(prefix string) ([]database.LiveRoute, map[int]database.ASN, error) {
|
||||||
// Return empty routes for now
|
// Return empty routes for now
|
||||||
return []database.LiveRoute{}, nil
|
return []database.LiveRoute{}, map[int]database.ASN{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetPrefixDetailsContext mock implementation with context support
|
// GetPrefixDetailsContext mock implementation with context support
|
||||||
func (m *mockStore) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]database.LiveRoute, error) {
|
func (m *mockStore) GetPrefixDetailsContext(
|
||||||
|
ctx context.Context, prefix string,
|
||||||
|
) ([]database.LiveRoute, map[int]database.ASN, error) {
|
||||||
return m.GetPrefixDetails(prefix)
|
return m.GetPrefixDetails(prefix)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -20,6 +20,8 @@ const (
|
|||||||
debugInterval = 60 * time.Second
|
debugInterval = 60 * time.Second
|
||||||
// bytesPerMB is bytes per megabyte
|
// bytesPerMB is bytes per megabyte
|
||||||
bytesPerMB = 1024 * 1024
|
bytesPerMB = 1024 * 1024
|
||||||
|
// failureExitCode is the exit status when RouteWatch stops on an error
|
||||||
|
failureExitCode = 1
|
||||||
)
|
)
|
||||||
|
|
||||||
// logDebugStats logs goroutine count and memory usage
|
// logDebugStats logs goroutine count and memory usage
|
||||||
@@ -83,6 +85,9 @@ func CLIEntry() {
|
|||||||
go func() {
|
go func() {
|
||||||
if err := rw.Run(ctx); err != nil {
|
if err := rw.Run(ctx); err != nil {
|
||||||
logger.Error("RouteWatch error", "error", err)
|
logger.Error("RouteWatch error", "error", err)
|
||||||
|
if err := shutdowner.Shutdown(fx.ExitCode(failureExitCode)); err != nil {
|
||||||
|
logger.Error("Failed to shutdown gracefully", "error", err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
|||||||
@@ -105,7 +105,24 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
|||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
defer h.mu.Unlock()
|
defer h.mu.Unlock()
|
||||||
|
|
||||||
// Process announcements
|
// Withdrawals are queued before announcements, because the batch keeps
|
||||||
|
// each peer's last update for a prefix: a message that lists a prefix as
|
||||||
|
// both withdrawn and announced keeps the route (RFC 4271 section 4.3).
|
||||||
|
for _, prefix := range msg.Withdrawals {
|
||||||
|
h.batch = append(h.batch, prefixUpdate{
|
||||||
|
prefix: canonicalPrefix(prefix),
|
||||||
|
originASN: originASN, // Use the originASN from path if available
|
||||||
|
peer: msg.Peer,
|
||||||
|
messageType: "withdrawal",
|
||||||
|
timestamp: timestamp,
|
||||||
|
path: msg.Path,
|
||||||
|
})
|
||||||
|
// Record withdrawal in metrics
|
||||||
|
if h.metrics != nil {
|
||||||
|
h.metrics.RecordWithdrawal()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
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{
|
||||||
@@ -123,22 +140,6 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Process withdrawals
|
|
||||||
for _, prefix := range msg.Withdrawals {
|
|
||||||
h.batch = append(h.batch, prefixUpdate{
|
|
||||||
prefix: canonicalPrefix(prefix),
|
|
||||||
originASN: originASN, // Use the originASN from path if available
|
|
||||||
peer: msg.Peer,
|
|
||||||
messageType: "withdrawal",
|
|
||||||
timestamp: timestamp,
|
|
||||||
path: msg.Path,
|
|
||||||
})
|
|
||||||
// Record withdrawal in metrics
|
|
||||||
if h.metrics != nil {
|
|
||||||
h.metrics.RecordWithdrawal()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check if we need to flush
|
// Check if we need to flush
|
||||||
if len(h.batch) >= prefixBatchSize {
|
if len(h.batch) >= prefixBatchSize {
|
||||||
h.flushBatchLocked()
|
h.flushBatchLocked()
|
||||||
@@ -179,14 +180,14 @@ func (h *PrefixHandler) flushBatchLocked() {
|
|||||||
startTime := time.Now()
|
startTime := time.Now()
|
||||||
batchSize := len(h.batch)
|
batchSize := len(h.batch)
|
||||||
|
|
||||||
// Group updates by prefix to deduplicate
|
// For each prefix and peer, keep only the last update the feed delivered.
|
||||||
// For each prefix, keep the latest update
|
// Updates from different peers for the same prefix are all kept. Arrival
|
||||||
prefixMap := make(map[string]prefixUpdate)
|
// order decides, not the timestamp: timestamps are whole seconds, so
|
||||||
|
// updates within one second carry the same one.
|
||||||
|
type prefixAndPeer struct{ prefix, peer string }
|
||||||
|
lastUpdates := make(map[prefixAndPeer]prefixUpdate)
|
||||||
for _, update := range h.batch {
|
for _, update := range h.batch {
|
||||||
key := update.prefix
|
lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update
|
||||||
if existing, ok := prefixMap[key]; !ok || update.timestamp.After(existing.timestamp) {
|
|
||||||
prefixMap[key] = update
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Collect routes to upsert and delete
|
// Collect routes to upsert and delete
|
||||||
@@ -196,7 +197,7 @@ func (h *PrefixHandler) flushBatchLocked() {
|
|||||||
// Collect unique prefixes to update
|
// Collect unique prefixes to update
|
||||||
prefixesToUpdate := make(map[string]time.Time)
|
prefixesToUpdate := make(map[string]time.Time)
|
||||||
|
|
||||||
for _, update := range prefixMap {
|
for _, update := range lastUpdates {
|
||||||
// Track prefix for both announcements and withdrawals
|
// Track prefix for both announcements and withdrawals
|
||||||
if _, exists := prefixesToUpdate[update.prefix]; !exists || update.timestamp.After(prefixesToUpdate[update.prefix]) {
|
if _, exists := prefixesToUpdate[update.prefix]; !exists || update.timestamp.After(prefixesToUpdate[update.prefix]) {
|
||||||
prefixesToUpdate[update.prefix] = update.timestamp
|
prefixesToUpdate[update.prefix] = update.timestamp
|
||||||
@@ -255,7 +256,7 @@ func (h *PrefixHandler) flushBatchLocked() {
|
|||||||
elapsed := time.Since(startTime)
|
elapsed := time.Since(startTime)
|
||||||
h.logger.Debug("Flushed prefix batch",
|
h.logger.Debug("Flushed prefix batch",
|
||||||
"batch_size", batchSize,
|
"batch_size", batchSize,
|
||||||
"unique_prefixes", len(prefixMap),
|
"unique_prefixes", len(prefixesToUpdate),
|
||||||
"success", successCount,
|
"success", successCount,
|
||||||
"duration_ms", elapsed.Milliseconds(),
|
"duration_ms", elapsed.Milliseconds(),
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -1,7 +1,9 @@
|
|||||||
package routewatch
|
package routewatch
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
"slices"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -72,3 +74,140 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestPrefixHandlerAppliesEachPeersLastUpdateInABatch runs updates for one
|
||||||
|
// prefix through the prefix handler into a real database. All updates carry the
|
||||||
|
// same timestamp, as updates within one second of the feed do. Within a batch,
|
||||||
|
// the last update each peer sent for the prefix decides whether that peer has a
|
||||||
|
// live route for it, and updates from different peers are all applied. The
|
||||||
|
// in-memory route count and prefix distribution must match the route table.
|
||||||
|
func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
|
||||||
|
const (
|
||||||
|
prefix = "198.51.100.0/24"
|
||||||
|
prefixMaskLength = 24
|
||||||
|
otherPeerIP = "2001:db8:ffff::2"
|
||||||
|
)
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
announce := func(peer string) *ristypes.RISMessage {
|
||||||
|
return &ristypes.RISMessage{
|
||||||
|
Peer: peer,
|
||||||
|
Path: ristypes.ASPath{testASNA, testASNB},
|
||||||
|
ParsedTimestamp: ts,
|
||||||
|
Announcements: []ristypes.RISAnnouncement{{NextHop: peer, Prefixes: []string{prefix}}},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
withdraw := func(peer string) *ristypes.RISMessage {
|
||||||
|
return &ristypes.RISMessage{Peer: peer, ParsedTimestamp: ts, Withdrawals: []string{prefix}}
|
||||||
|
}
|
||||||
|
announceAndWithdraw := func(peer string) *ristypes.RISMessage {
|
||||||
|
msg := announce(peer)
|
||||||
|
msg.Withdrawals = []string{prefix}
|
||||||
|
|
||||||
|
return msg
|
||||||
|
}
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
// Each inner list is handled and then flushed as one batch.
|
||||||
|
batches [][]*ristypes.RISMessage
|
||||||
|
// The peers left with a live route for the prefix, by peer IP.
|
||||||
|
wantPeers []string
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "two peers announce the prefix",
|
||||||
|
batches: [][]*ristypes.RISMessage{{announce(testPeerIP), announce(otherPeerIP)}},
|
||||||
|
wantPeers: []string{testPeerIP, otherPeerIP},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "two peers withdraw the prefix",
|
||||||
|
batches: [][]*ristypes.RISMessage{
|
||||||
|
{announce(testPeerIP)},
|
||||||
|
{announce(otherPeerIP)},
|
||||||
|
{withdraw(testPeerIP), withdraw(otherPeerIP)},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "a peer announces and then withdraws the prefix",
|
||||||
|
batches: [][]*ristypes.RISMessage{{announce(testPeerIP), withdraw(testPeerIP)}},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "a peer withdraws and then re-announces the prefix",
|
||||||
|
batches: [][]*ristypes.RISMessage{
|
||||||
|
{announce(testPeerIP)},
|
||||||
|
{withdraw(testPeerIP), announce(testPeerIP)},
|
||||||
|
},
|
||||||
|
wantPeers: []string{testPeerIP},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
// RFC 4271 section 4.3: the announcement counts, not the withdrawal.
|
||||||
|
name: "one message announces and withdraws the prefix",
|
||||||
|
batches: [][]*ristypes.RISMessage{
|
||||||
|
{announce(testPeerIP)},
|
||||||
|
{announceAndWithdraw(testPeerIP)},
|
||||||
|
},
|
||||||
|
wantPeers: []string{testPeerIP},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(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()}
|
||||||
|
for _, batch := range tt.batches {
|
||||||
|
for _, msg := range batch {
|
||||||
|
h.HandleMessage(msg)
|
||||||
|
}
|
||||||
|
h.mu.Lock()
|
||||||
|
h.flushBatchLocked()
|
||||||
|
h.mu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
routes, _, err := db.GetPrefixDetailsContext(ctx, prefix)
|
||||||
|
if err != nil && !errors.Is(err, database.ErrNoRoute) {
|
||||||
|
t.Fatalf("GetPrefixDetailsContext: %v", err)
|
||||||
|
}
|
||||||
|
var gotPeers []string
|
||||||
|
for _, route := range routes {
|
||||||
|
gotPeers = append(gotPeers, route.PeerIP)
|
||||||
|
}
|
||||||
|
if !slices.Equal(gotPeers, tt.wantPeers) {
|
||||||
|
t.Errorf("peers with a live route for %s = %v, want %v", prefix, gotPeers, tt.wantPeers)
|
||||||
|
}
|
||||||
|
|
||||||
|
stats, err := db.GetStatsContext(ctx)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetStatsContext: %v", err)
|
||||||
|
}
|
||||||
|
tableRoutesV4, _, err := db.GetLiveRouteCountsContext(ctx)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetLiveRouteCountsContext: %v", err)
|
||||||
|
}
|
||||||
|
if stats.IPv4Routes != len(tt.wantPeers) || tableRoutesV4 != len(tt.wantPeers) {
|
||||||
|
t.Errorf("IPv4 live routes: %d in memory, %d in the table, want %d",
|
||||||
|
stats.IPv4Routes, tableRoutesV4, len(tt.wantPeers))
|
||||||
|
}
|
||||||
|
|
||||||
|
var wantDistribution []database.PrefixDistribution
|
||||||
|
if len(tt.wantPeers) > 0 {
|
||||||
|
wantDistribution = []database.PrefixDistribution{{MaskLength: prefixMaskLength, Count: 1}}
|
||||||
|
}
|
||||||
|
queryDistributionV4, _, err := db.GetPrefixDistributionContext(ctx)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("GetPrefixDistributionContext: %v", err)
|
||||||
|
}
|
||||||
|
if !slices.Equal(stats.IPv4PrefixDistribution, wantDistribution) ||
|
||||||
|
!slices.Equal(queryDistributionV4, wantDistribution) {
|
||||||
|
t.Errorf("IPv4 prefix distribution: %v in memory, %v from the table, want %v",
|
||||||
|
stats.IPv4PrefixDistribution, queryDistributionV4, wantDistribution)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+23
-16
@@ -7,6 +7,7 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"net/netip"
|
||||||
"runtime"
|
"runtime"
|
||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
@@ -748,10 +749,17 @@ func (s *Server) handlePrefixDetailJSON() http.HandlerFunc {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Combine prefix and length into CIDR notation
|
// Look the prefix up in the text form net/netip prints, the form the
|
||||||
prefix := prefixParam + "/" + lenParam
|
// tables store, so that any valid way of writing it finds the route.
|
||||||
|
parsed, err := netip.ParsePrefix(prefixParam + "/" + lenParam)
|
||||||
|
if err != nil {
|
||||||
|
writeJSONError(w, http.StatusBadRequest, err.Error())
|
||||||
|
|
||||||
routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
return
|
||||||
|
}
|
||||||
|
prefix := parsed.Masked().String()
|
||||||
|
|
||||||
|
routes, _, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, database.ErrNoRoute) {
|
if errors.Is(err, database.ErrNoRoute) {
|
||||||
writeJSONError(w, http.StatusNotFound, err.Error())
|
writeJSONError(w, http.StatusNotFound, err.Error())
|
||||||
@@ -912,10 +920,17 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Combine prefix and length into CIDR notation
|
// Look the prefix up in the text form net/netip prints, the form the
|
||||||
prefix := prefixParam + "/" + lenParam
|
// tables store, so that any valid way of writing it finds the route.
|
||||||
|
parsed, err := netip.ParsePrefix(prefixParam + "/" + lenParam)
|
||||||
|
if err != nil {
|
||||||
|
http.Error(w, "Invalid prefix", http.StatusBadRequest)
|
||||||
|
|
||||||
routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
return
|
||||||
|
}
|
||||||
|
prefix := parsed.Masked().String()
|
||||||
|
|
||||||
|
routes, originASNs, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, database.ErrNoRoute) {
|
if errors.Is(err, database.ErrNoRoute) {
|
||||||
http.Error(w, "Prefix not found", http.StatusNotFound)
|
http.Error(w, "Prefix not found", http.StatusNotFound)
|
||||||
@@ -937,18 +952,10 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
|
|||||||
originMap := make(map[int]*ASNInfo)
|
originMap := make(map[int]*ASNInfo)
|
||||||
for _, route := range routes {
|
for _, route := range routes {
|
||||||
if _, exists := originMap[route.OriginASN]; !exists {
|
if _, exists := originMap[route.OriginASN]; !exists {
|
||||||
// Get AS info from database
|
|
||||||
asInfo, _, _ := s.db.GetASDetailsContext(r.Context(), route.OriginASN)
|
|
||||||
handle := ""
|
|
||||||
description := ""
|
|
||||||
if asInfo != nil {
|
|
||||||
handle = asInfo.Handle
|
|
||||||
description = asInfo.Description
|
|
||||||
}
|
|
||||||
originMap[route.OriginASN] = &ASNInfo{
|
originMap[route.OriginASN] = &ASNInfo{
|
||||||
ASN: route.OriginASN,
|
ASN: route.OriginASN,
|
||||||
Handle: handle,
|
Handle: originASNs[route.OriginASN].Handle,
|
||||||
Description: description,
|
Description: originASNs[route.OriginASN].Description,
|
||||||
PeerCount: 0,
|
PeerCount: 0,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,10 +3,13 @@ package server
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"html"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"runtime"
|
"runtime"
|
||||||
"slices"
|
"slices"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -15,6 +18,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"
|
||||||
|
"git.eeqj.de/sneak/routewatch/pkg/asinfo"
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -148,6 +152,124 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestPrefixDetailHandlersParsePrefix checks that both prefix pages find a
|
||||||
|
// stored route whatever valid text form the URL writes its prefix in, and
|
||||||
|
// answer 400 for a prefix that does not parse. The tables store the form
|
||||||
|
// net/netip prints. Before the fix the URL text was looked up as typed, so
|
||||||
|
// "2001:DB8:0:0::/32" was not found although "2001:db8::/32" was stored, and
|
||||||
|
// "foo/24" answered 500 (https://git.eeqj.de/sneak/routewatch/issues/71).
|
||||||
|
func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
|
||||||
|
db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("database.New: %v", err)
|
||||||
|
}
|
||||||
|
t.Cleanup(func() { _ = db.Close() })
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*database.LiveRoute{
|
||||||
|
{
|
||||||
|
ID: uuid.New(), Prefix: "192.0.2.0/24", MaskLength: 24, IPVersion: 4,
|
||||||
|
OriginASN: 64500, PeerIP: "198.51.100.1", ASPath: []int{64500}, NextHop: "198.51.100.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)
|
||||||
|
}
|
||||||
|
|
||||||
|
s := New(db, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
path string
|
||||||
|
want int
|
||||||
|
}{
|
||||||
|
{"/api/v1/prefix/foo/24", http.StatusBadRequest},
|
||||||
|
{"/prefix/foo/24", http.StatusBadRequest},
|
||||||
|
{"/api/v1/prefix/2001:DB8:0:0::/32", http.StatusOK},
|
||||||
|
{"/prefix/2001:DB8:0:0::/32", http.StatusOK},
|
||||||
|
{"/api/v1/prefix/192.0.2.1/24", http.StatusOK},
|
||||||
|
{"/prefix/192.0.2.1/24", http.StatusOK},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
s.router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, tt.path, nil))
|
||||||
|
if rec.Code != tt.want {
|
||||||
|
t.Errorf("%s: status %d, want %d; body %s", tt.path, rec.Code, tt.want, rec.Body)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// failingASDetailsDB wraps a real store and fails every AS details lookup.
|
||||||
|
type failingASDetailsDB struct {
|
||||||
|
database.Store
|
||||||
|
}
|
||||||
|
|
||||||
|
func (failingASDetailsDB) GetASDetailsContext(_ context.Context, _ int) (*database.ASN, []database.LiveRoute, error) {
|
||||||
|
return nil, nil, errors.New("AS details lookup failed")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestPrefixDetailShowsOriginNames checks that the prefix page shows each
|
||||||
|
// origin AS's handle and description without looking up the AS's details,
|
||||||
|
// which also read every route the AS originates. Before the fix the page did
|
||||||
|
// that lookup once per origin, so with it failing the names were missing
|
||||||
|
// (https://git.eeqj.de/sneak/routewatch/issues/67). The AS path column shows
|
||||||
|
// handles too, so the test looks for the origin list's "(handle)" form.
|
||||||
|
func TestPrefixDetailShowsOriginNames(t *testing.T) {
|
||||||
|
db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("database.New: %v", err)
|
||||||
|
}
|
||||||
|
t.Cleanup(func() { _ = db.Close() })
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
origins := []int{15169, 13335}
|
||||||
|
for _, asn := range origins {
|
||||||
|
if _, err := db.GetOrCreateASN(asn, ts); err != nil {
|
||||||
|
t.Fatalf("GetOrCreateASN(%d): %v", asn, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*database.LiveRoute{
|
||||||
|
{
|
||||||
|
ID: uuid.New(), Prefix: "192.0.2.0/24", MaskLength: 24, IPVersion: 4,
|
||||||
|
OriginASN: origins[0], PeerIP: "198.51.100.1", ASPath: []int{origins[0]}, NextHop: "198.51.100.1",
|
||||||
|
LastUpdated: ts,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
ID: uuid.New(), Prefix: "192.0.2.0/24", MaskLength: 24, IPVersion: 4,
|
||||||
|
OriginASN: origins[1], PeerIP: "198.51.100.2", ASPath: []int{origins[1]}, NextHop: "198.51.100.2",
|
||||||
|
LastUpdated: ts,
|
||||||
|
},
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
s := New(failingASDetailsDB{db}, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
s.router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/prefix/192.0.2.0/24", nil))
|
||||||
|
if rec.Code != http.StatusOK {
|
||||||
|
t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
|
||||||
|
}
|
||||||
|
|
||||||
|
page := rec.Body.String()
|
||||||
|
for _, asn := range origins {
|
||||||
|
info, ok := asinfo.Get(asn)
|
||||||
|
if !ok {
|
||||||
|
t.Fatalf("AS%d is not in the AS info data", asn)
|
||||||
|
}
|
||||||
|
if want := "(" + html.EscapeString(info.Handle) + ")"; !strings.Contains(page, want) {
|
||||||
|
t.Errorf("page does not show AS%d's handle %q", asn, want)
|
||||||
|
}
|
||||||
|
if want := html.EscapeString(info.Description); !strings.Contains(page, want) {
|
||||||
|
t.Errorf("page does not show AS%d's description %q", asn, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// 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 {
|
||||||
|
|||||||
@@ -3,6 +3,8 @@ package server
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
"strconv"
|
"strconv"
|
||||||
"time"
|
"time"
|
||||||
@@ -53,9 +55,12 @@ func New(db database.Store, streamer *streamer.Streamer, logger *logger.Logger,
|
|||||||
return s
|
return s
|
||||||
}
|
}
|
||||||
|
|
||||||
// Start starts the HTTP server
|
// Start listens on the configured port and serves HTTP in the background.
|
||||||
|
// It returns an error if it cannot listen, for example because the port is
|
||||||
|
// already in use.
|
||||||
func (s *Server) Start() error {
|
func (s *Server) Start() error {
|
||||||
const (
|
const (
|
||||||
|
network = "tcp"
|
||||||
readHeaderTimeout = 40 * time.Second
|
readHeaderTimeout = 40 * time.Second
|
||||||
readTimeout = 60 * time.Second
|
readTimeout = 60 * time.Second
|
||||||
writeTimeout = 60 * time.Second
|
writeTimeout = 60 * time.Second
|
||||||
@@ -73,10 +78,15 @@ func (s *Server) Start() error {
|
|||||||
|
|
||||||
s.logger.Info("Starting HTTP server", "port", s.port, "addr", s.srv.Addr)
|
s.logger.Info("Starting HTTP server", "port", s.port, "addr", s.srv.Addr)
|
||||||
|
|
||||||
// Start in goroutine but log when actually listening
|
listener, err := net.Listen(network, s.srv.Addr)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("starting HTTP server: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
s.logger.Info("HTTP server listening", "addr", listener.Addr().String())
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
s.logger.Info("HTTP server listening", "addr", s.srv.Addr)
|
if err := s.srv.Serve(listener); err != nil && err != http.ErrServerClosed {
|
||||||
if err := s.srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
|
||||||
s.logger.Error("HTTP server error", "error", err)
|
s.logger.Error("HTTP server error", "error", err)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|||||||
@@ -0,0 +1,33 @@
|
|||||||
|
package server
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"net"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/config"
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/metrics"
|
||||||
|
"git.eeqj.de/sneak/routewatch/internal/streamer"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestStartFailsWhenPortIsHeld holds a port with a listener of its own and
|
||||||
|
// checks that Start on that port returns an error. Before the fix Start
|
||||||
|
// listened in a goroutine, only logged the failure and returned nil, so the
|
||||||
|
// daemon kept running without its web server.
|
||||||
|
func TestStartFailsWhenPortIsHeld(t *testing.T) {
|
||||||
|
held, err := net.Listen("tcp", ":0")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to open the test listener: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = held.Close() }()
|
||||||
|
|
||||||
|
port := held.Addr().(*net.TCPAddr).Port
|
||||||
|
log := logger.New()
|
||||||
|
s := New(nil, streamer.New(log, metrics.New()), log, &config.Config{Port: port})
|
||||||
|
defer func() { _ = s.Stop(context.Background()) }()
|
||||||
|
|
||||||
|
if err := s.Start(); err == nil {
|
||||||
|
t.Fatalf("Start returned nil while port %d was held by another listener", port)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -283,6 +283,7 @@
|
|||||||
<div class="origin-item">
|
<div class="origin-item">
|
||||||
<a href="/as/{{.ASN}}" class="as-link">AS{{.ASN}}</a>
|
<a href="/as/{{.ASN}}" class="as-link">AS{{.ASN}}</a>
|
||||||
{{if .Handle}} ({{.Handle}}){{end}}
|
{{if .Handle}} ({{.Handle}}){{end}}
|
||||||
|
{{if .Description}} {{.Description}}{{end}}
|
||||||
<span style="color: #7f8c8d; margin-left: 10px;">{{.PeerCount}} peer{{if ne .PeerCount 1}}s{{end}}</span>
|
<span style="color: #7f8c8d; margin-left: 10px;">{{.PeerCount}} peer{{if ne .PeerCount 1}}s{{end}}</span>
|
||||||
</div>
|
</div>
|
||||||
{{end}}
|
{{end}}
|
||||||
|
|||||||
Reference in New Issue
Block a user