The AS page and /api/v1/as/{asn} grouped every route the AS originates
by prefix, reading one table row per peer and prefix and sorting them
all. They now take the AS's prefixes one at a time: one search of a new
index on origin_asn and prefix, which replaces the origin_asn index and
is written exactly when it was, and one search of the prefix and
last_updated index for the time of the prefix's newest route. Neither
reads the table. For a prefix more than one AS originates, that time is
now the newest route's from any origin. A hand-run benchmark reads an
AS of 5,000 prefixes from 500 peers.
Model: opus-5-5
This commit was merged in pull request #129.
This commit is contained in:
@@ -1673,8 +1673,9 @@ func (d *Database) GetASDetails(asn int) (*ASN, []LiveRoute, error) {
|
||||
return d.GetASDetailsContext(context.Background(), asn)
|
||||
}
|
||||
|
||||
// GetASDetailsContext returns detailed information about an AS including
|
||||
// prefixes with context support
|
||||
// GetASDetailsContext returns the AS and the prefixes it originates, the IPv4
|
||||
// ones first, each once with the time of its newest route. It makes two index
|
||||
// searches per prefix, however many peers carry the prefix.
|
||||
func (d *Database) GetASDetailsContext(
|
||||
ctx context.Context, asn int,
|
||||
) (*ASN, []LiveRoute, error) {
|
||||
@@ -1701,27 +1702,12 @@ func (d *Database) GetASDetailsContext(
|
||||
asnInfo.Handle = handle.String
|
||||
asnInfo.Description = description.String
|
||||
|
||||
// Get prefixes announced by this AS from both tables
|
||||
queryV4 := `
|
||||
SELECT prefix, mask_length, MAX(last_updated) as last_updated
|
||||
FROM live_routes_v4
|
||||
WHERE origin_asn = ?
|
||||
GROUP BY prefix, mask_length
|
||||
`
|
||||
|
||||
allPrefixes, err := d.originPrefixes(ctx, queryV4, asn, ipVersionV4)
|
||||
allPrefixes, err := d.originPrefixes(ctx, asn, ipVersionV4)
|
||||
if err != nil {
|
||||
return &asnInfo, nil, err
|
||||
}
|
||||
|
||||
queryV6 := `
|
||||
SELECT prefix, mask_length, MAX(last_updated) as last_updated
|
||||
FROM live_routes_v6
|
||||
WHERE origin_asn = ?
|
||||
GROUP BY prefix, mask_length
|
||||
`
|
||||
|
||||
prefixesV6, err := d.originPrefixes(ctx, queryV6, asn, ipVersionV6)
|
||||
prefixesV6, err := d.originPrefixes(ctx, asn, ipVersionV6)
|
||||
if err != nil {
|
||||
return &asnInfo, allPrefixes, err
|
||||
}
|
||||
@@ -2452,66 +2438,87 @@ func (d *Database) scanRouteTimestamp(
|
||||
}
|
||||
}
|
||||
|
||||
// originPrefixes runs query, which selects the prefix, mask length and latest
|
||||
// last_updated of every prefix asn originates in the live route table of
|
||||
// ipVersion. Rows that fail to scan or whose timestamp does not parse are
|
||||
// logged and skipped.
|
||||
// The AS page lists the prefixes an AS originates one at a time, each in two
|
||||
// index searches that read no table row: one of the origin_asn and prefix
|
||||
// index for the AS's first prefix after the previous one, and one of the
|
||||
// prefix and last_updated index for the time of its newest route. Neither
|
||||
// reads more for a prefix that more peers carry.
|
||||
const (
|
||||
nextOriginPrefixV4SQL = `SELECT prefix FROM live_routes_v4
|
||||
WHERE origin_asn = ? AND prefix > ? ORDER BY prefix LIMIT 1`
|
||||
newestRouteTimeV4SQL = `SELECT last_updated FROM live_routes_v4
|
||||
WHERE prefix = ? ORDER BY last_updated DESC LIMIT 1`
|
||||
|
||||
nextOriginPrefixV6SQL = `SELECT prefix FROM live_routes_v6
|
||||
WHERE origin_asn = ? AND prefix > ? ORDER BY prefix LIMIT 1`
|
||||
newestRouteTimeV6SQL = `SELECT last_updated FROM live_routes_v6
|
||||
WHERE prefix = ? ORDER BY last_updated DESC LIMIT 1`
|
||||
)
|
||||
|
||||
// originPrefixes returns every prefix asn originates in the live route table
|
||||
// of ipVersion, in text order, each with its mask length and the time of the
|
||||
// prefix's newest route, whichever AS that route names as origin. A prefix
|
||||
// whose routes were all withdrawn between the two searches is left out.
|
||||
func (d *Database) originPrefixes(
|
||||
ctx context.Context, query string, asn, ipVersion int,
|
||||
ctx context.Context, asn, ipVersion int,
|
||||
) ([]LiveRoute, error) {
|
||||
family := ipv4Family
|
||||
family, nextSQL, newestSQL := ipv4Family, nextOriginPrefixV4SQL,
|
||||
newestRouteTimeV4SQL
|
||||
if ipVersion == ipVersionV6 {
|
||||
family = ipv6Family
|
||||
family, nextSQL, newestSQL = ipv6Family, nextOriginPrefixV6SQL,
|
||||
newestRouteTimeV6SQL
|
||||
}
|
||||
|
||||
rows, err := d.db.QueryContext(ctx, query, asn)
|
||||
nextStmt, err := d.db.PrepareContext(ctx, nextSQL)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to query %s prefixes: %w", family, err)
|
||||
return nil, fmt.Errorf("failed to prepare %s prefix query: %w",
|
||||
family, err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
defer func() { _ = nextStmt.Close() }()
|
||||
|
||||
newestStmt, err := d.db.PrepareContext(ctx, newestSQL)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to prepare %s route time query: %w",
|
||||
family, err)
|
||||
}
|
||||
defer func() { _ = newestStmt.Close() }()
|
||||
|
||||
var routes []LiveRoute
|
||||
|
||||
for rows.Next() {
|
||||
var (
|
||||
route LiveRoute
|
||||
lastUpdatedStr string
|
||||
)
|
||||
previous := "" // every prefix sorts after the empty string
|
||||
|
||||
for {
|
||||
var prefix string
|
||||
|
||||
err := nextStmt.QueryRowContext(ctx, asn, previous).Scan(&prefix)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return routes, nil
|
||||
}
|
||||
|
||||
err := rows.Scan(&route.Prefix, &route.MaskLength, &lastUpdatedStr)
|
||||
if err != nil {
|
||||
d.logger.Error("Failed to scan "+family+" prefix row",
|
||||
"error", err, "asn", asn)
|
||||
return nil, fmt.Errorf("failed to list %s prefixes: %w", family, err)
|
||||
}
|
||||
|
||||
previous = prefix
|
||||
route := LiveRoute{Prefix: prefix, IPVersion: ipVersion, OriginASN: asn}
|
||||
|
||||
err = newestStmt.QueryRowContext(ctx, prefix).Scan(&route.LastUpdated)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Parse the timestamp string
|
||||
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05-07:00",
|
||||
lastUpdatedStr)
|
||||
if err != nil {
|
||||
// Try without timezone
|
||||
route.LastUpdated, err = time.Parse("2006-01-02 15:04:05",
|
||||
lastUpdatedStr)
|
||||
if err != nil {
|
||||
d.logger.Error("Failed to parse timestamp",
|
||||
"error", err, "timestamp", lastUpdatedStr)
|
||||
return nil, fmt.Errorf("failed to query route time of %s: %w",
|
||||
prefix, err)
|
||||
}
|
||||
|
||||
continue
|
||||
}
|
||||
route.MaskLength, err = prefixMaskLength(prefix)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
route.OriginASN = asn
|
||||
route.IPVersion = ipVersion
|
||||
routes = append(routes, route)
|
||||
}
|
||||
|
||||
err = rows.Err()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read %s prefixes: %w", family, err)
|
||||
}
|
||||
|
||||
return routes, nil
|
||||
}
|
||||
|
||||
// firstPrefixesByLength returns the first n prefixes, in text order, with the
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"net/netip"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -1214,6 +1215,138 @@ func TestLiveRouteUpdateSetsOriginWhenItChanges(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestGetASDetailsListsEachPrefixOnce stores two IPv4 prefixes and an IPv6
|
||||
// prefix an AS originates, two of them from two peers at different times, and
|
||||
// a prefix another AS originates. The AS details list each of the AS's
|
||||
// prefixes once, IPv4 first and in text order, with its mask length and the
|
||||
// time of its newest route.
|
||||
func TestGetASDetailsListsEachPrefixOnce(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
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() }()
|
||||
|
||||
const asn, otherASN = 64500, 64501
|
||||
|
||||
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||
|
||||
_, err = db.GetOrCreateASN(asn, ts)
|
||||
if err != nil {
|
||||
t.Fatalf("GetOrCreateASN: %v", err)
|
||||
}
|
||||
|
||||
v4Older := mkV4Route(t, "192.0.2.0/24", asn, ts)
|
||||
v4Newer := mkV4Route(t, "192.0.2.0/24", asn, ts.Add(time.Minute))
|
||||
v4Newer.PeerIP = secondV4PeerIP
|
||||
v4Single := mkV4Route(t, "198.51.100.128/25", asn, ts)
|
||||
v6Newer := mkV6Route(testV6Prefix, asn, ts.Add(time.Minute))
|
||||
v6Older := mkV6Route(testV6Prefix, asn, ts)
|
||||
v6Older.PeerIP = secondV6PeerIP
|
||||
other := mkV4Route(t, "203.0.113.0/24", otherASN, ts)
|
||||
|
||||
err = db.UpsertLiveRouteBatch([]*LiveRoute{
|
||||
v4Older, v4Newer, v4Single, v6Newer, v6Older, other,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||
}
|
||||
|
||||
_, prefixes, err := db.GetASDetails(asn)
|
||||
if err != nil {
|
||||
t.Fatalf("GetASDetails: %v", err)
|
||||
}
|
||||
|
||||
want := []*LiveRoute{v4Newer, v4Single, v6Newer}
|
||||
|
||||
listedOnce := slices.EqualFunc(prefixes, want,
|
||||
func(got LiveRoute, want *LiveRoute) bool {
|
||||
return got.Prefix == want.Prefix &&
|
||||
got.MaskLength == want.MaskLength &&
|
||||
got.IPVersion == want.IPVersion &&
|
||||
got.OriginASN == asn &&
|
||||
got.LastUpdated.Equal(want.LastUpdated)
|
||||
})
|
||||
if !listedOnce {
|
||||
t.Errorf("GetASDetails listed %+v, want the prefixes and times of %+v",
|
||||
prefixes, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestASPrefixQueriesUseCoveringIndexes checks the plan of each query that
|
||||
// lists the prefixes an AS originates: a search of a covering index and no
|
||||
// temporary B-tree, so the list reads one index entry per prefix, not one
|
||||
// table row per route. Grouping the AS's routes by prefix read every route's
|
||||
// row and sorted them in a temporary B-tree
|
||||
// (https://git.eeqj.de/sneak/routewatch/issues/104).
|
||||
func TestASPrefixQueriesUseCoveringIndexes(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
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() }()
|
||||
|
||||
tests := []struct {
|
||||
query string
|
||||
args []any
|
||||
index string
|
||||
}{
|
||||
{nextOriginPrefixV4SQL, []any{0, ""}, "idx_live_routes_v4_origin_prefix"},
|
||||
{newestRouteTimeV4SQL, []any{""}, "idx_live_routes_v4_prefix_last_updated"},
|
||||
{nextOriginPrefixV6SQL, []any{0, ""}, "idx_live_routes_v6_origin_prefix"},
|
||||
{newestRouteTimeV6SQL, []any{""}, "idx_live_routes_v6_prefix_last_updated"},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
plan := queryPlan(t, db, tt.query, tt.args...)
|
||||
|
||||
if !strings.Contains(plan, "USING COVERING INDEX "+tt.index+" ") ||
|
||||
strings.Contains(plan, "TEMP B-TREE") {
|
||||
t.Errorf("plan of %s:\n%s\nwant a search of covering index %s "+
|
||||
"and no temporary B-tree", tt.query, plan, tt.index)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// queryPlan returns the steps of EXPLAIN QUERY PLAN for query, one per line.
|
||||
func queryPlan(t *testing.T, db *Database, query string, args ...any) string {
|
||||
t.Helper()
|
||||
|
||||
rows, err := db.db.QueryContext(t.Context(), "EXPLAIN QUERY PLAN "+query,
|
||||
args...)
|
||||
if err != nil {
|
||||
t.Fatalf("EXPLAIN QUERY PLAN %s: %v", query, err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var steps []string
|
||||
|
||||
for rows.Next() {
|
||||
var (
|
||||
id, parent, unused int
|
||||
detail string
|
||||
)
|
||||
|
||||
err := rows.Scan(&id, &parent, &unused, &detail)
|
||||
if err != nil {
|
||||
t.Fatalf("reading the plan of %s: %v", query, err)
|
||||
}
|
||||
|
||||
steps = append(steps, detail)
|
||||
}
|
||||
|
||||
err = rows.Err()
|
||||
if err != nil {
|
||||
t.Fatalf("reading the plan of %s: %v", query, err)
|
||||
}
|
||||
|
||||
return strings.Join(steps, "\n")
|
||||
}
|
||||
|
||||
// TestUpdatePrefixesBatchKeepsFirstSeen writes a prefix, then a batch with it
|
||||
// and a new prefix. Only the new prefix is inserted and counted, and the known
|
||||
// one keeps its first_seen time.
|
||||
@@ -1518,3 +1651,66 @@ func writeBenchBatch(
|
||||
b.Fatalf("UpdatePrefixesBatch: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Parameters for BenchmarkGetASDetailsWithFiveHundredPeers: the AS originates
|
||||
// benchASPrefixes prefixes and every one of benchASPeers peers has a route for
|
||||
// each, 2.5 million routes in all.
|
||||
const (
|
||||
benchASPrefixes = 5000
|
||||
benchASPeers = 500
|
||||
)
|
||||
|
||||
// BenchmarkGetASDetailsWithFiveHundredPeers times the read behind the AS page
|
||||
// for an AS that originates 5,000 IPv4 prefixes, each from 500 peers. It
|
||||
// builds that database first, one peer at a time, which takes some minutes and
|
||||
// about a GB of disk under TMPDIR. It is run by hand, not by make test, for
|
||||
// example in the test image:
|
||||
//
|
||||
// go test -run '^$' -bench GetASDetails -benchtime 10x ./internal/database/
|
||||
func BenchmarkGetASDetailsWithFiveHundredPeers(b *testing.B) {
|
||||
db, err := New(&config.Config{StateDir: b.TempDir()}, logger.New())
|
||||
if err != nil {
|
||||
b.Fatalf("failed to create database: %v", err)
|
||||
}
|
||||
defer func() { _ = db.Close() }()
|
||||
|
||||
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||
|
||||
_, err = db.GetOrCreateASN(benchFirstASN, ts)
|
||||
if err != nil {
|
||||
b.Fatalf("GetOrCreateASN: %v", err)
|
||||
}
|
||||
|
||||
for peer := range benchASPeers {
|
||||
peerIP := fmt.Sprintf("198.18.%d.%d", peer/256, peer%256)
|
||||
|
||||
routes := make([]*LiveRoute, 0, benchASPrefixes)
|
||||
for prefix := range benchASPrefixes {
|
||||
routes = append(routes, &LiveRoute{
|
||||
Prefix: fmt.Sprintf("%d.%d.%d.0/%d",
|
||||
1+prefix>>16, (prefix>>8)%256, prefix%256, benchMaskLength),
|
||||
MaskLength: benchMaskLength,
|
||||
IPVersion: ipVersionV4,
|
||||
OriginASN: benchFirstASN,
|
||||
PeerIP: peerIP,
|
||||
ASPath: []int{benchFirstASN + 1 + peer, benchFirstASN},
|
||||
NextHop: peerIP,
|
||||
LastUpdated: ts,
|
||||
})
|
||||
}
|
||||
|
||||
writeBenchBatch(b, db, routes, nil, ts)
|
||||
}
|
||||
|
||||
for b.Loop() {
|
||||
_, prefixes, err := db.GetASDetails(benchFirstASN)
|
||||
if err != nil {
|
||||
b.Fatalf("GetASDetails: %v", err)
|
||||
}
|
||||
|
||||
if len(prefixes) != benchASPrefixes {
|
||||
b.Fatalf("GetASDetails listed %d prefixes, want %d",
|
||||
len(prefixes), benchASPrefixes)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -128,7 +128,7 @@ CREATE TABLE IF NOT EXISTS live_routes_v6 (
|
||||
-- prefix, last_updated: a prefix's routes, for the prefix page and its API
|
||||
-- answer, the IP lookup and its peer count, and the check whether a prefix
|
||||
-- still has a live route; and a prefix's newest route for the prefix-length
|
||||
-- pages, in one search whatever the number of peers.
|
||||
-- pages and the AS page, in one search whatever the number of peers.
|
||||
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_prefix_last_updated ON live_routes_v4(prefix, last_updated);
|
||||
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_prefix_last_updated ON live_routes_v6(prefix, last_updated);
|
||||
-- mask_length: the prefix-length pages' random sample, the row id range of a
|
||||
@@ -139,9 +139,10 @@ CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_length ON live_routes_v6(mask
|
||||
-- prefixes in text order, and the prefix distribution read at startup.
|
||||
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_v6_mask_prefix ON live_routes_v6(mask_length, prefix);
|
||||
-- origin_asn: the prefixes an AS originates, for the AS page and its API answer.
|
||||
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_v6_origin_asn ON live_routes_v6(origin_asn);
|
||||
-- origin_asn, prefix: the prefixes an AS originates, each in one search
|
||||
-- whatever the number of peers, for the AS page and its API answer.
|
||||
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_prefix ON live_routes_v4(origin_asn, prefix);
|
||||
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_origin_prefix ON live_routes_v6(origin_asn, prefix);
|
||||
-- last_updated: the oldest and newest route times in /api/v1/stats and
|
||||
-- /status.json.
|
||||
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated);
|
||||
|
||||
Reference in New Issue
Block a user