Compare commits
1
Commits
next
..
49d2f4ede8
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
49d2f4ede8 |
@@ -142,16 +142,7 @@ and origin AS can have two kept routes, one or none.
|
|||||||
- `/prefix/{prefix}/{length}`: a prefix, such as `/prefix/192.0.2.0/24`, with
|
- `/prefix/{prefix}/{length}`: a prefix, such as `/prefix/192.0.2.0/24`, with
|
||||||
its origin ASes and every peer that carries it. Only the kept routes show an
|
its origin ASes and every peer that carries it. Only the kept routes show an
|
||||||
AS path, a next hop and a time. The next hop is the router the peer sends the
|
AS path, a next hop and a time. The next hop is the router the peer sends the
|
||||||
prefix's traffic to. RouteWatch records the peer itself as the next hop. The
|
prefix's traffic to. RouteWatch records the peer itself as the next hop.
|
||||||
page also lists the prefix's origin changes, newest first. An origin change is
|
|
||||||
an AS starting to announce the prefix, when its first peer carries it, or
|
|
||||||
stopping, when its last peer no longer does. The page shows at most the newest
|
|
||||||
5,000. Each row shows:
|
|
||||||
- the time
|
|
||||||
- announced or withdrawn
|
|
||||||
- the origin AS
|
|
||||||
- the peer count: how many peers carry the prefix just after the change,
|
|
||||||
with any origin AS
|
|
||||||
- `/prefixlength/{length}`: up to 500 IPv4 prefixes with that mask length,
|
- `/prefixlength/{length}`: up to 500 IPv4 prefixes with that mask length,
|
||||||
picked at random. Each shows its origin AS and the age of its newest kept
|
picked at random. Each shows its origin AS and the age of its newest kept
|
||||||
route. When peers give a prefix different origin ASes, the page shows the one
|
route. When peers give a prefix different origin ASes, the page shows the one
|
||||||
@@ -525,8 +516,7 @@ The ceilings add up to 3.7 GiB:
|
|||||||
pairs, so it takes about 1 GB. The rest of the heap takes up to about 0.5 GiB,
|
pairs, so it takes about 1 GB. The rest of the heap takes up to about 0.5 GiB,
|
||||||
most of it the handlers' batches and queues (see below). The remaining 1 GiB
|
most of it the handlers' batches and queues (see below). The remaining 1 GiB
|
||||||
is room for the garbage collector, which frees memory the program no longer
|
is room for the garbage collector, which frees memory the program no longer
|
||||||
uses. With much less, it would run almost all the time. The routing table also
|
uses. With much less, it would run almost all the time.
|
||||||
keeps at most 100,000 origin changes waiting to be written, about 8 MB.
|
|
||||||
- SQLite: 1 GiB, set in the code. The connection that does every write has a 512
|
- SQLite: 1 GiB, set in the code. The connection that does every write has a 512
|
||||||
MiB page cache: the database pages it keeps in memory from one write to the
|
MiB page cache: the database pages it keeps in memory from one write to the
|
||||||
next. The `routes` table of a full routing table takes about 450 MB, so its
|
next. The `routes` table of a full routing table takes about 450 MB, so its
|
||||||
@@ -577,9 +567,6 @@ What happens at each limit:
|
|||||||
and after the held-back messages of every peer whose session went down before
|
and after the held-back messages of every peer whose session went down before
|
||||||
it. It holds at most 150,000 such messages, about 150 MiB. At that number it
|
it. It holds at most 150,000 such messages, about 150 MiB. At that number it
|
||||||
takes nothing more off its queue until it has applied one of them.
|
takes nothing more off its queue until it has applied one of them.
|
||||||
- Waiting origin changes: the routing table keeps at most 100,000 origin changes
|
|
||||||
waiting to be written. Past that, it drops the oldest, and the prefix handler
|
|
||||||
logs how many. The prefix page never shows a dropped origin change.
|
|
||||||
|
|
||||||
To replace the image's Go soft limit, set `GOMEMLIMIT` in the environment, for
|
To replace the image's Go soft limit, set `GOMEMLIMIT` in the environment, for
|
||||||
example `-e GOMEMLIMIT=3GiB`. A limit close to the routing table's size makes
|
example `-e GOMEMLIMIT=3GiB`. A limit close to the routing table's size makes
|
||||||
@@ -633,9 +620,8 @@ The parts:
|
|||||||
messages. It gives each message to the handlers that want it, through one
|
messages. It gives each message to the handlers that want it, through one
|
||||||
queue per handler.
|
queue per handler.
|
||||||
- The prefix handler keeps the routing table in memory current. Every second it
|
- The prefix handler keeps the routing table in memory current. Every second it
|
||||||
writes the table's changes to the `routes` table and the table's origin
|
writes the table's changes to the `routes` table of the database. It also
|
||||||
changes to the `origin_changes` table of the database. It also stores each new
|
stores each new prefix with the time it was first seen.
|
||||||
prefix with the time it was first seen.
|
|
||||||
- The AS handler records every AS seen in an AS path in the `asns` table, with
|
- The AS handler records every AS seen in an AS path in the `asns` table, with
|
||||||
the times it was first and last seen.
|
the times it was first and last seen.
|
||||||
- The peer handler records every peer in the `bgp_peers` table.
|
- The peer handler records every peer in the `bgp_peers` table.
|
||||||
@@ -652,14 +638,11 @@ The parts:
|
|||||||
- The database maintainer checkpoints the database every 5 seconds: it folds the
|
- The database maintainer checkpoints the database every 5 seconds: it folds the
|
||||||
write-ahead log, where SQLite first writes changes, into the database file. It
|
write-ahead log, where SQLite first writes changes, into the database file. It
|
||||||
reclaims free space every 10 minutes. Every hour it refreshes the statistics
|
reclaims free space every 10 minutes. Every hour it refreshes the statistics
|
||||||
SQLite uses to plan its queries, and prunes the `origin_changes` table. A
|
SQLite uses to plan its queries.
|
||||||
prune keeps each prefix's rows from the last 72 hours, or its newest 500,
|
|
||||||
whichever keeps more.
|
|
||||||
- The HTTP server serves the pages and the API. Every route it shows comes from
|
- The HTTP server serves the pages and the API. Every route it shows comes from
|
||||||
the routing table in memory. WHOIS details, first-seen times and origin
|
the routing table in memory. WHOIS details and first-seen times come from the
|
||||||
changes come from the database. AS names come from the database or from the AS
|
database. AS names come from the database or from the AS list built into
|
||||||
list built into RouteWatch. The AS name search reads the AS list built into
|
RouteWatch. The AS name search reads the AS list built into RouteWatch.
|
||||||
RouteWatch.
|
|
||||||
|
|
||||||
### The routing table
|
### The routing table
|
||||||
|
|
||||||
@@ -770,8 +753,6 @@ What a stop can lose:
|
|||||||
changed pairs. A full-table peer is a peer that carries the whole routing
|
changed pairs. A full-table peer is a peer that carries the whole routing
|
||||||
table. Its session going down changes about a million pairs, which take about
|
table. Its session going down changes about a million pairs, which take about
|
||||||
a minute to write. A stop in that minute leaves the rest unwritten.
|
a minute to write. A stop in that minute leaves the rest unwritten.
|
||||||
- The prefix handler's last write of origin changes takes at most 20,000. The
|
|
||||||
rest are lost.
|
|
||||||
- A withdrawal, or a session going down, that is lost this way leaves its routes
|
- A withdrawal, or a session going down, that is lost this way leaves its routes
|
||||||
in the `routes` table. The next start reads them back. They stay until the
|
in the `routes` table. The next start reads them back. They stay until the
|
||||||
peer sends another update for the prefix, or its session goes down again.
|
peer sends another update for the prefix, or its session goes down again.
|
||||||
@@ -828,10 +809,6 @@ routes(prefix, origin_asn, peers,
|
|||||||
-- The peers of the routes table, each with its number in the peer sets
|
-- The peers of the routes table, each with its number in the peer sets
|
||||||
route_peers(number PRIMARY KEY, address UNIQUE)
|
route_peers(number PRIMARY KEY, address UNIQUE)
|
||||||
|
|
||||||
-- The origin changes of the routing table: each time a prefix and origin AS
|
|
||||||
-- gained its first peer or lost its last
|
|
||||||
origin_changes(id PRIMARY KEY, prefix, origin_asn, announced, peer_count, time)
|
|
||||||
|
|
||||||
peerings(id PRIMARY KEY, as_a, as_b, first_seen, last_seen,
|
peerings(id PRIMARY KEY, as_a, as_b, first_seen, last_seen,
|
||||||
UNIQUE(as_a, as_b))
|
UNIQUE(as_a, as_b))
|
||||||
bgp_peers(id PRIMARY KEY, peer_ip UNIQUE, peer_asn, first_seen, last_seen,
|
bgp_peers(id PRIMARY KEY, peer_ip UNIQUE, peer_asn, first_seen, last_seen,
|
||||||
|
|||||||
@@ -41,8 +41,9 @@ https://git.eeqj.de/sneak/routewatch/issues/225, also on `next`, takes the peers
|
|||||||
whose sessions go down together out of the table in one walk. A 24-hour run of
|
whose sessions go down together out of the table in one walk. A 24-hour run of
|
||||||
`9bd0e91` on the live feed is checking whether any handler still drops updates.
|
`9bd0e91` on the live feed is checking whether any handler still drops updates.
|
||||||
|
|
||||||
One question waits for sneak: which fields the IP lookup answer gains
|
Two questions wait for sneak: which history the prefix page keeps
|
||||||
(https://git.eeqj.de/sneak/routewatch/issues/203).
|
(https://git.eeqj.de/sneak/routewatch/issues/161), and which fields the IP
|
||||||
|
lookup answer gains (https://git.eeqj.de/sneak/routewatch/issues/203).
|
||||||
|
|
||||||
The order of the project's remaining work is kept on
|
The order of the project's remaining work is kept on
|
||||||
https://git.eeqj.de/sneak/routewatch/issues/86.
|
https://git.eeqj.de/sneak/routewatch/issues/86.
|
||||||
@@ -56,19 +57,6 @@ problem the run finds.
|
|||||||
|
|
||||||
# Completed Steps
|
# Completed Steps
|
||||||
|
|
||||||
- 2026-11-05: the prefix page lists the prefix's origin changes, newest first.
|
|
||||||
The routing table records each change, and the prefix handler writes them to a
|
|
||||||
new `origin_changes` table. The database maintainer prunes each prefix to its
|
|
||||||
newest 500 rows or its rows from the last 72 hours, whichever keeps more. The
|
|
||||||
page shows at most the newest 5,000. Each row shows:
|
|
||||||
- the time
|
|
||||||
- announced or withdrawn
|
|
||||||
- the origin AS
|
|
||||||
- the peer count
|
|
||||||
|
|
||||||
A row's peer count is how many peers carry the prefix just after the change,
|
|
||||||
with any origin AS (closes #161).
|
|
||||||
|
|
||||||
- 2026-11-03: the streamer drops an update for a handler only when that
|
- 2026-11-03: the streamer drops an update for a handler only when that
|
||||||
handler's queue is full. It used to drop a random share of the updates once
|
handler's queue is full. It used to drop a random share of the updates once
|
||||||
the queue was half full. The prefix handler's queue now holds 100,000
|
the queue was half full. The prefix handler's queue now holds 100,000
|
||||||
|
|||||||
@@ -1,7 +1,6 @@
|
|||||||
// Package database provides SQLite storage for BGP routing data: ASNs,
|
// Package database provides SQLite storage for BGP routing data: ASNs,
|
||||||
// prefixes, announcements, peerings, BGP peers and origin changes. Its routes
|
// prefixes, announcements, peerings and BGP peers. Its routes table holds a
|
||||||
// table holds a copy of the whole routing table that internal/routetable
|
// copy of the whole routing table that internal/routetable keeps in memory.
|
||||||
// keeps in memory.
|
|
||||||
package database
|
package database
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -56,14 +55,6 @@ type Store interface {
|
|||||||
WriteRouteChanges(changes []routetable.Change, newPeers map[string]uint) error
|
WriteRouteChanges(changes []routetable.Change, newPeers map[string]uint) error
|
||||||
ReadRouteTable(ctx context.Context) (*routetable.Table, error)
|
ReadRouteTable(ctx context.Context) (*routetable.Table, error)
|
||||||
|
|
||||||
// Origin change operations: the prefix and origin AS pairs' first and
|
|
||||||
// last peers, as the routing table records them.
|
|
||||||
WriteOriginChanges(changes []routetable.OriginChange) error
|
|
||||||
GetOriginChanges(
|
|
||||||
ctx context.Context, prefix netip.Prefix, limit int,
|
|
||||||
) ([]routetable.OriginChange, error)
|
|
||||||
PruneOriginChanges(ctx context.Context) error
|
|
||||||
|
|
||||||
// ASN WHOIS operations
|
// ASN WHOIS operations
|
||||||
GetNextStaleASN(ctx context.Context, staleThreshold time.Duration) (int, error)
|
GetNextStaleASN(ctx context.Context, staleThreshold time.Duration) (int, error)
|
||||||
UpdateASNWHOIS(ctx context.Context, update *ASNWHOISUpdate) error
|
UpdateASNWHOIS(ctx context.Context, update *ASNWHOISUpdate) error
|
||||||
|
|||||||
@@ -1,172 +0,0 @@
|
|||||||
package database
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"fmt"
|
|
||||||
"net/netip"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/routetable"
|
|
||||||
)
|
|
||||||
|
|
||||||
// minOriginChangesPerPrefix is how many of a prefix's newest origin changes
|
|
||||||
// the prune always keeps, whatever their age.
|
|
||||||
const minOriginChangesPerPrefix = 500
|
|
||||||
|
|
||||||
// originChangesRetention is how long the prune keeps an origin change: rows
|
|
||||||
// older than this are deleted, unless they are among the prefix's newest
|
|
||||||
// minOriginChangesPerPrefix.
|
|
||||||
const originChangesRetention = 72 * time.Hour
|
|
||||||
|
|
||||||
// The statements of the origin_changes table (schema.sql).
|
|
||||||
const (
|
|
||||||
insertOriginChangeSQL = `INSERT INTO origin_changes
|
|
||||||
(prefix, origin_asn, announced, peer_count, time)
|
|
||||||
VALUES (?, ?, ?, ?, ?)`
|
|
||||||
originChangesSQL = `SELECT prefix, origin_asn, announced, peer_count, time
|
|
||||||
FROM origin_changes WHERE prefix = ? ORDER BY id DESC LIMIT ?`
|
|
||||||
prunePrefixesSQL = `SELECT prefix FROM origin_changes
|
|
||||||
GROUP BY prefix HAVING COUNT(*) > ?`
|
|
||||||
pruneOriginChangesSQL = `DELETE FROM origin_changes
|
|
||||||
WHERE prefix = ? AND time < ? AND id < (
|
|
||||||
SELECT id FROM origin_changes WHERE prefix = ?
|
|
||||||
ORDER BY id DESC LIMIT 1 OFFSET ?)`
|
|
||||||
)
|
|
||||||
|
|
||||||
// WriteOriginChanges writes, in one transaction, the origin changes the
|
|
||||||
// routing table handed over. When it fails, it writes nothing; the caller then
|
|
||||||
// writes them again.
|
|
||||||
func (d *Database) WriteOriginChanges(changes []routetable.OriginChange) error {
|
|
||||||
if len(changes) == 0 {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
d.lock("WriteOriginChanges")
|
|
||||||
defer d.unlock()
|
|
||||||
|
|
||||||
tx, err := d.beginTx()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to begin transaction: %w", err)
|
|
||||||
}
|
|
||||||
defer d.rollback(tx)
|
|
||||||
|
|
||||||
ctx := context.Background()
|
|
||||||
|
|
||||||
stmt, err := tx.PrepareContext(ctx, insertOriginChangeSQL)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to prepare origin change insert: %w", err)
|
|
||||||
}
|
|
||||||
defer func() { _ = stmt.Close() }()
|
|
||||||
|
|
||||||
for _, change := range changes {
|
|
||||||
_, err := stmt.ExecContext(ctx, change.Prefix.String(),
|
|
||||||
change.OriginASN, change.Announced, change.PeerCount,
|
|
||||||
change.Time.Unix())
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to write origin change %s: %w",
|
|
||||||
change.Prefix, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
err = tx.Commit()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to commit transaction: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetOriginChanges returns at most limit of prefix's origin changes, newest
|
|
||||||
// first.
|
|
||||||
func (d *Database) GetOriginChanges(
|
|
||||||
ctx context.Context, prefix netip.Prefix, limit int,
|
|
||||||
) ([]routetable.OriginChange, error) {
|
|
||||||
rows, err := d.db.QueryContext(ctx, originChangesSQL,
|
|
||||||
prefix.Masked().String(), limit)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("failed to read origin changes: %w", err)
|
|
||||||
}
|
|
||||||
defer func() { _ = rows.Close() }()
|
|
||||||
|
|
||||||
var changes []routetable.OriginChange
|
|
||||||
|
|
||||||
for rows.Next() {
|
|
||||||
var (
|
|
||||||
change routetable.OriginChange
|
|
||||||
prefixText string
|
|
||||||
timeSec int64
|
|
||||||
)
|
|
||||||
|
|
||||||
err := rows.Scan(&prefixText, &change.OriginASN, &change.Announced,
|
|
||||||
&change.PeerCount, &timeSec)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("failed to read an origin change: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
change.Prefix, err = netip.ParsePrefix(prefixText)
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("failed to read origin change %s: %w",
|
|
||||||
prefixText, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
change.Time = time.Unix(timeSec, 0).UTC()
|
|
||||||
changes = append(changes, change)
|
|
||||||
}
|
|
||||||
|
|
||||||
err = rows.Err()
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("failed to read origin changes: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return changes, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// PruneOriginChanges deletes, for each prefix with more than
|
|
||||||
// minOriginChangesPerPrefix rows, the rows older than originChangesRetention
|
|
||||||
// that are not among its newest minOriginChangesPerPrefix. It finds those
|
|
||||||
// prefixes on a reading connection and deletes one prefix per write, so no
|
|
||||||
// write holds the lock for long.
|
|
||||||
func (d *Database) PruneOriginChanges(ctx context.Context) error {
|
|
||||||
rows, err := d.db.QueryContext(ctx, prunePrefixesSQL,
|
|
||||||
minOriginChangesPerPrefix)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to find prefixes to prune: %w", err)
|
|
||||||
}
|
|
||||||
defer func() { _ = rows.Close() }()
|
|
||||||
|
|
||||||
var prefixes []string
|
|
||||||
|
|
||||||
for rows.Next() {
|
|
||||||
var prefix string
|
|
||||||
|
|
||||||
err := rows.Scan(&prefix)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to read a prefix to prune: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
prefixes = append(prefixes, prefix)
|
|
||||||
}
|
|
||||||
|
|
||||||
err = rows.Err()
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to find prefixes to prune: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
cutoff := time.Now().Add(-originChangesRetention).Unix()
|
|
||||||
|
|
||||||
for _, prefix := range prefixes {
|
|
||||||
d.lock("PruneOriginChanges")
|
|
||||||
|
|
||||||
_, err := d.writer.ExecContext(ctx, pruneOriginChangesSQL,
|
|
||||||
prefix, cutoff, prefix, minOriginChangesPerPrefix-1)
|
|
||||||
|
|
||||||
d.unlock()
|
|
||||||
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("failed to prune origin changes of %s: %w",
|
|
||||||
prefix, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
@@ -1,217 +0,0 @@
|
|||||||
package database
|
|
||||||
|
|
||||||
import (
|
|
||||||
"net/netip"
|
|
||||||
"reflect"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/routetable"
|
|
||||||
)
|
|
||||||
|
|
||||||
// TestOriginChangesWriteThenRead writes three origin changes and reads them
|
|
||||||
// back newest first, honouring the limit.
|
|
||||||
func TestOriginChangesWriteThenRead(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
db := newRoutesTestDatabase(t)
|
|
||||||
prefix := netip.MustParsePrefix(tablePrefixV4)
|
|
||||||
|
|
||||||
changes := []routetable.OriginChange{
|
|
||||||
{Prefix: prefix, OriginASN: asA, Announced: true, PeerCount: 1,
|
|
||||||
Time: tableTime(10)},
|
|
||||||
{Prefix: prefix, OriginASN: asA, Announced: false, PeerCount: 0,
|
|
||||||
Time: tableTime(20)},
|
|
||||||
{Prefix: prefix, OriginASN: asB, Announced: true, PeerCount: 1,
|
|
||||||
Time: tableTime(30)},
|
|
||||||
}
|
|
||||||
|
|
||||||
err := db.WriteOriginChanges(changes)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("WriteOriginChanges: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
got, err := db.GetOriginChanges(t.Context(), prefix, 2)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("GetOriginChanges: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
want := []routetable.OriginChange{
|
|
||||||
{Prefix: prefix, OriginASN: asB, Announced: true, PeerCount: 1,
|
|
||||||
Time: tableTime(30)},
|
|
||||||
{Prefix: prefix, OriginASN: asA, Announced: false, PeerCount: 0,
|
|
||||||
Time: tableTime(20)},
|
|
||||||
}
|
|
||||||
|
|
||||||
if !reflect.DeepEqual(got, want) {
|
|
||||||
t.Errorf("GetOriginChanges = %+v, want %+v", got, want)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOriginChangesPrune checks that the prune keeps, for a prefix with more
|
|
||||||
// than 500 rows, its newest 500 rows and every row newer than 72 hours, and
|
|
||||||
// deletes the rest. It checks which rows survive by their times, not only how
|
|
||||||
// many.
|
|
||||||
func TestOriginChangesPrune(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
db := newRoutesTestDatabase(t)
|
|
||||||
oldPrefix := netip.MustParsePrefix(tablePrefixV4)
|
|
||||||
straddling := netip.MustParsePrefix(tablePrefixShortV4)
|
|
||||||
busy := netip.MustParsePrefix(tablePrefixBusyV4)
|
|
||||||
other := netip.MustParsePrefix("10.20.0.0/16")
|
|
||||||
|
|
||||||
// Whole seconds, so the stored times round-trip exactly.
|
|
||||||
now := time.Now().Truncate(time.Second)
|
|
||||||
|
|
||||||
// 600 rows for oldPrefix, all older than 72 hours, oldest first. The
|
|
||||||
// prune keeps the newest 500 and deletes the oldest 100.
|
|
||||||
changes := rows(oldPrefix, now.Add(-100*time.Hour), 600)
|
|
||||||
// 600 rows for straddling: 400 older than 72 hours, then 200 newer. The
|
|
||||||
// prune keeps all 200 newer and the newest 300 of the 400 older.
|
|
||||||
changes = append(changes, rows(straddling, now.Add(-100*time.Hour), 400)...)
|
|
||||||
changes = append(changes, rows(straddling, now.Add(-71*time.Hour), 200)...)
|
|
||||||
// 100 rows older than 72 hours and minOriginChangesPerPrefix+100 newer
|
|
||||||
// rows for busy. The prune keeps all the newer rows and none of the older,
|
|
||||||
// because the newer rows alone are more than minOriginChangesPerPrefix.
|
|
||||||
changes = append(changes, rows(busy, now.Add(-100*time.Hour), 100)...)
|
|
||||||
changes = append(changes, rows(busy, now.Add(-71*time.Hour),
|
|
||||||
minOriginChangesPerPrefix+100)...)
|
|
||||||
// One row for other, which the prune leaves alone.
|
|
||||||
changes = append(changes, rows(other, now.Add(-100*time.Hour), 1)...)
|
|
||||||
|
|
||||||
err := db.WriteOriginChanges(changes)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("WriteOriginChanges: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
err = db.PruneOriginChanges(t.Context())
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("PruneOriginChanges: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// The newest row survives, and the oldest survivor is the 100th oldest,
|
|
||||||
// so a prune that kept the oldest 500 instead would fail.
|
|
||||||
checkSurvivors(t, db, oldPrefix, 500,
|
|
||||||
now.Add(-100*time.Hour).Add(599*time.Minute),
|
|
||||||
now.Add(-100*time.Hour).Add(100*time.Minute))
|
|
||||||
|
|
||||||
// The newest of the newer-than-72-hours rows survives, and the oldest
|
|
||||||
// survivor is the 100th oldest of the older-than-72-hours rows.
|
|
||||||
checkSurvivors(t, db, straddling, 500,
|
|
||||||
now.Add(-71*time.Hour).Add(199*time.Minute),
|
|
||||||
now.Add(-100*time.Hour).Add(100*time.Minute))
|
|
||||||
|
|
||||||
// Every newer-than-72-hours row survives, so the oldest survivor is the
|
|
||||||
// first of them and the newest is minOriginChangesPerPrefix+99 minutes
|
|
||||||
// after it.
|
|
||||||
checkSurvivors(t, db, busy, minOriginChangesPerPrefix+100,
|
|
||||||
now.Add(-71*time.Hour).
|
|
||||||
Add(time.Duration(minOriginChangesPerPrefix+99)*time.Minute),
|
|
||||||
now.Add(-71*time.Hour))
|
|
||||||
|
|
||||||
gotOther, err := db.GetOriginChanges(t.Context(), other, 10)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("GetOriginChanges: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(gotOther) != 1 {
|
|
||||||
t.Errorf("GetOriginChanges(other) = %d rows, want 1", len(gotOther))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// rows returns n origin changes for prefix, at times from plus one minute per
|
|
||||||
// step, oldest first.
|
|
||||||
func rows(prefix netip.Prefix, from time.Time, n int) []routetable.OriginChange {
|
|
||||||
changes := make([]routetable.OriginChange, 0, n)
|
|
||||||
|
|
||||||
for i := range n {
|
|
||||||
changes = append(changes, routetable.OriginChange{
|
|
||||||
Prefix: prefix, OriginASN: asA, Announced: true, PeerCount: 1,
|
|
||||||
Time: from.Add(time.Duration(i) * time.Minute),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
return changes
|
|
||||||
}
|
|
||||||
|
|
||||||
// survivorTimes returns the times of prefix's rows after the prune, newest
|
|
||||||
// first.
|
|
||||||
func survivorTimes(
|
|
||||||
t *testing.T, db *Database, prefix netip.Prefix, limit int,
|
|
||||||
) []time.Time {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
rows, err := db.GetOriginChanges(t.Context(), prefix, limit)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("GetOriginChanges: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
times := make([]time.Time, len(rows))
|
|
||||||
for i, row := range rows {
|
|
||||||
times[i] = row.Time
|
|
||||||
}
|
|
||||||
|
|
||||||
return times
|
|
||||||
}
|
|
||||||
|
|
||||||
// checkSurvivors checks that prefix kept exactly wantCount rows, with the
|
|
||||||
// newest survivor at wantNewest and the oldest at wantOldest.
|
|
||||||
func checkSurvivors(
|
|
||||||
t *testing.T, db *Database, prefix netip.Prefix,
|
|
||||||
wantCount int, wantNewest, wantOldest time.Time,
|
|
||||||
) {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
times := survivorTimes(t, db, prefix, wantCount+1)
|
|
||||||
if len(times) != wantCount {
|
|
||||||
t.Errorf("GetOriginChanges(%s) = %d rows, want %d",
|
|
||||||
prefix, len(times), wantCount)
|
|
||||||
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
if !times[0].Equal(wantNewest) {
|
|
||||||
t.Errorf("the newest surviving row's time = %v, want %v",
|
|
||||||
times[0], wantNewest)
|
|
||||||
}
|
|
||||||
|
|
||||||
if !times[wantCount-1].Equal(wantOldest) {
|
|
||||||
t.Errorf("the oldest surviving row's time = %v, want %v",
|
|
||||||
times[wantCount-1], wantOldest)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOriginChangesSurviveRouteDeletion checks that deleting a routes row
|
|
||||||
// leaves the prefix's origin_changes rows.
|
|
||||||
func TestOriginChangesSurviveRouteDeletion(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
db := newRoutesTestDatabase(t)
|
|
||||||
prefix := netip.MustParsePrefix(tablePrefixV4)
|
|
||||||
|
|
||||||
err := db.WriteOriginChanges([]routetable.OriginChange{{
|
|
||||||
Prefix: prefix, OriginASN: asA, Announced: true, PeerCount: 1,
|
|
||||||
Time: tableTime(10),
|
|
||||||
}})
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("WriteOriginChanges: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Write a route, then delete it.
|
|
||||||
table := routetable.New(routetable.Params{})
|
|
||||||
announceTo(table, tablePrefixV4, asA, peerA, 10)
|
|
||||||
writeTableChanges(t, db, table)
|
|
||||||
|
|
||||||
table.Withdraw(prefix, peerA)
|
|
||||||
writeTableChanges(t, db, table)
|
|
||||||
|
|
||||||
got, err := db.GetOriginChanges(t.Context(), prefix, 10)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("GetOriginChanges: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(got) != 1 {
|
|
||||||
t.Errorf("GetOriginChanges = %d rows, want 1", len(got))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -34,7 +34,6 @@ const (
|
|||||||
|
|
||||||
tablePrefixV4 = "10.1.2.0/24"
|
tablePrefixV4 = "10.1.2.0/24"
|
||||||
tablePrefixShortV4 = "10.9.0.0/16"
|
tablePrefixShortV4 = "10.9.0.0/16"
|
||||||
tablePrefixBusyV4 = "10.10.0.0/16"
|
|
||||||
tablePrefixV6 = "2001:db8:1::/48"
|
tablePrefixV6 = "2001:db8:1::/48"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -120,19 +120,3 @@ CREATE TABLE IF NOT EXISTS route_peers (
|
|||||||
number INTEGER PRIMARY KEY,
|
number INTEGER PRIMARY KEY,
|
||||||
address TEXT NOT NULL UNIQUE
|
address TEXT NOT NULL UNIQUE
|
||||||
);
|
);
|
||||||
|
|
||||||
-- The origin changes of the routing table: each time a prefix and origin AS
|
|
||||||
-- gained its first peer or lost its last. The prefix handler writes them, and
|
|
||||||
-- the database maintainer prunes each prefix to its newest 500 rows or its
|
|
||||||
-- rows from the last 72 hours, whichever keeps more. time is whole seconds
|
|
||||||
-- since 1970, UTC.
|
|
||||||
CREATE TABLE IF NOT EXISTS origin_changes (
|
|
||||||
id INTEGER PRIMARY KEY,
|
|
||||||
prefix TEXT NOT NULL,
|
|
||||||
origin_asn INTEGER NOT NULL,
|
|
||||||
announced BOOLEAN NOT NULL,
|
|
||||||
peer_count INTEGER NOT NULL,
|
|
||||||
time INTEGER NOT NULL
|
|
||||||
);
|
|
||||||
|
|
||||||
CREATE INDEX IF NOT EXISTS idx_origin_changes_prefix ON origin_changes(prefix);
|
|
||||||
@@ -1,103 +0,0 @@
|
|||||||
package routetable
|
|
||||||
|
|
||||||
import (
|
|
||||||
"net/netip"
|
|
||||||
"slices"
|
|
||||||
"time"
|
|
||||||
)
|
|
||||||
|
|
||||||
// MaxWaitingOriginChanges is the most origin changes the table keeps waiting
|
|
||||||
// for the prefix handler to write. The oldest change past it is dropped and
|
|
||||||
// counted.
|
|
||||||
const MaxWaitingOriginChanges = 100000
|
|
||||||
|
|
||||||
// OriginChange is one change of a prefix and origin AS pair: the origin AS
|
|
||||||
// gained its first peer, or lost its last.
|
|
||||||
type OriginChange struct {
|
|
||||||
Prefix netip.Prefix
|
|
||||||
OriginASN int
|
|
||||||
// Announced says the origin AS gained its first peer. When false, it
|
|
||||||
// lost its last peer.
|
|
||||||
Announced bool
|
|
||||||
// PeerCount is how many peers carry the prefix just after the change,
|
|
||||||
// counting every origin AS.
|
|
||||||
PeerCount int
|
|
||||||
// Time is when RouteWatch applied the change.
|
|
||||||
Time time.Time
|
|
||||||
}
|
|
||||||
|
|
||||||
// recordOriginChange records that the pair of prefix and asn gained its first
|
|
||||||
// peer, when announced, or lost its last, when not. The caller holds the lock.
|
|
||||||
// peerCount is how many peers carry the prefix just after the change.
|
|
||||||
func (t *Table) recordOriginChange(
|
|
||||||
prefix netip.Prefix, asn int, announced bool, peerCount int,
|
|
||||||
) {
|
|
||||||
change := OriginChange{
|
|
||||||
Prefix: prefix,
|
|
||||||
OriginASN: asn,
|
|
||||||
Announced: announced,
|
|
||||||
PeerCount: peerCount,
|
|
||||||
Time: t.now(),
|
|
||||||
}
|
|
||||||
|
|
||||||
t.originChanges = append(t.originChanges, change)
|
|
||||||
t.trimOriginChanges()
|
|
||||||
}
|
|
||||||
|
|
||||||
// TakeOriginChanges returns at most limit of the origin changes that wait to
|
|
||||||
// be written, oldest first, and how many were dropped since the last call. It
|
|
||||||
// then forgets what it returned.
|
|
||||||
func (t *Table) TakeOriginChanges(limit int) ([]OriginChange, int) {
|
|
||||||
t.mu.Lock()
|
|
||||||
defer t.mu.Unlock()
|
|
||||||
|
|
||||||
n := min(limit, len(t.originChanges))
|
|
||||||
changes := slices.Clone(t.originChanges[:n])
|
|
||||||
t.originChanges = t.originChanges[n:]
|
|
||||||
|
|
||||||
dropped := t.originChangesDropped
|
|
||||||
t.originChangesDropped = 0
|
|
||||||
|
|
||||||
return changes, dropped
|
|
||||||
}
|
|
||||||
|
|
||||||
// PutBackOriginChanges records again what one TakeOriginChanges call returned,
|
|
||||||
// for a caller that could not write it. The next TakeOriginChanges returns
|
|
||||||
// those changes first, before the ones that arrived since. The oldest changes
|
|
||||||
// past the cap are dropped and counted.
|
|
||||||
func (t *Table) PutBackOriginChanges(changes []OriginChange) {
|
|
||||||
t.mu.Lock()
|
|
||||||
defer t.mu.Unlock()
|
|
||||||
|
|
||||||
t.originChanges = append(changes, t.originChanges...)
|
|
||||||
t.trimOriginChanges()
|
|
||||||
}
|
|
||||||
|
|
||||||
// trimOriginChanges drops the oldest origin changes past the cap and counts
|
|
||||||
// them. The caller holds the lock.
|
|
||||||
func (t *Table) trimOriginChanges() {
|
|
||||||
if extra := len(t.originChanges) - MaxWaitingOriginChanges; extra > 0 {
|
|
||||||
t.originChangesDropped += extra
|
|
||||||
t.originChanges = t.originChanges[extra:]
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// OriginChangeCount returns how many origin changes wait for TakeOriginChanges.
|
|
||||||
func (t *Table) OriginChangeCount() int {
|
|
||||||
t.mu.RLock()
|
|
||||||
defer t.mu.RUnlock()
|
|
||||||
|
|
||||||
return len(t.originChanges)
|
|
||||||
}
|
|
||||||
|
|
||||||
// prefixPeerCount returns how many peers carry prefix, counting every origin
|
|
||||||
// AS. The caller holds the lock.
|
|
||||||
func (t *Table) prefixPeerCount(prefix netip.Prefix) int {
|
|
||||||
count := 0
|
|
||||||
|
|
||||||
for _, o := range t.origins(prefix) {
|
|
||||||
count += int(o.peers.Count())
|
|
||||||
}
|
|
||||||
|
|
||||||
return count
|
|
||||||
}
|
|
||||||
@@ -1,302 +0,0 @@
|
|||||||
package routetable_test
|
|
||||||
|
|
||||||
import (
|
|
||||||
"math"
|
|
||||||
"net/netip"
|
|
||||||
"reflect"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/bits-and-blooms/bitset"
|
|
||||||
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/routetable"
|
|
||||||
)
|
|
||||||
|
|
||||||
// originChange returns the origin change the table records for prefix and asn
|
|
||||||
// at the clock's time, announced or withdrawn, with peerCount peers.
|
|
||||||
func originChange(
|
|
||||||
prefix netip.Prefix, asn int, announced bool, peerCount int, at time.Time,
|
|
||||||
) routetable.OriginChange {
|
|
||||||
return routetable.OriginChange{
|
|
||||||
Prefix: prefix,
|
|
||||||
OriginASN: asn,
|
|
||||||
Announced: announced,
|
|
||||||
PeerCount: peerCount,
|
|
||||||
Time: at,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOriginChangesRecordAnnouncedAndWithdrawn checks that a prefix and origin
|
|
||||||
// AS pair records an announced row when it gains its first peer, nothing for a
|
|
||||||
// second peer, and a withdrawn row when it loses its last peer.
|
|
||||||
func TestOriginChangesRecordAnnouncedAndWithdrawn(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
now := at(0)
|
|
||||||
table := routetable.New(routetable.Params{
|
|
||||||
Now: func() time.Time { return now },
|
|
||||||
})
|
|
||||||
prefix := netip.MustParsePrefix(prefixV4)
|
|
||||||
|
|
||||||
table.Announce(prefix, asA, routeOf(peerA, asA, 10))
|
|
||||||
table.Announce(prefix, asA, routeOf(peerB, asA, 20))
|
|
||||||
|
|
||||||
checkOriginChanges(t, table, []routetable.OriginChange{
|
|
||||||
originChange(prefix, asA, true, 1, now),
|
|
||||||
})
|
|
||||||
|
|
||||||
table.Withdraw(prefix, peerA)
|
|
||||||
table.Withdraw(prefix, peerB)
|
|
||||||
|
|
||||||
checkOriginChanges(t, table, []routetable.OriginChange{
|
|
||||||
originChange(prefix, asA, false, 0, now),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOriginChangesRecordAPeerMovingOrigin checks that a peer moving from one
|
|
||||||
// origin AS to another records a withdrawn row for the old one and an
|
|
||||||
// announced row for the new one.
|
|
||||||
func TestOriginChangesRecordAPeerMovingOrigin(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
now := at(0)
|
|
||||||
table := routetable.New(routetable.Params{
|
|
||||||
Now: func() time.Time { return now },
|
|
||||||
})
|
|
||||||
prefix := netip.MustParsePrefix(prefixV4)
|
|
||||||
|
|
||||||
table.Announce(prefix, asA, routeOf(peerA, asA, 10))
|
|
||||||
table.Announce(prefix, asB, routeOf(peerA, asB, 20))
|
|
||||||
|
|
||||||
checkOriginChanges(t, table, []routetable.OriginChange{
|
|
||||||
originChange(prefix, asA, true, 1, now),
|
|
||||||
originChange(prefix, asA, false, 0, now),
|
|
||||||
originChange(prefix, asB, true, 1, now),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOriginChangesRecordPeersDown checks that PeersDown records a withdrawn
|
|
||||||
// row for each prefix and origin AS pair it empties.
|
|
||||||
func TestOriginChangesRecordPeersDown(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
now := at(0)
|
|
||||||
table := routetable.New(routetable.Params{
|
|
||||||
Now: func() time.Time { return now },
|
|
||||||
})
|
|
||||||
v4 := netip.MustParsePrefix(prefixV4)
|
|
||||||
v6 := netip.MustParsePrefix(prefixV6)
|
|
||||||
|
|
||||||
table.Announce(v4, asA, routeOf(peerA, asA, 10))
|
|
||||||
table.Announce(v6, asA, routeOf(peerA, asA, 10))
|
|
||||||
table.Announce(v4, asB, routeOf(peerB, asB, 20))
|
|
||||||
|
|
||||||
table.PeersDown([]string{peerA})
|
|
||||||
|
|
||||||
// peerA carried v4 and v6 with asA; both pairs are emptied. peerB still
|
|
||||||
// carries v4 with asB, so that pair is not.
|
|
||||||
checkOriginChanges(t, table, []routetable.OriginChange{
|
|
||||||
originChange(v4, asA, true, 1, now),
|
|
||||||
originChange(v6, asA, true, 1, now),
|
|
||||||
originChange(v4, asB, true, 2, now),
|
|
||||||
originChange(v4, asA, false, 1, now),
|
|
||||||
originChange(v6, asA, false, 0, now),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOriginChangesLoadRecordsNothing checks that Load records no origin
|
|
||||||
// change.
|
|
||||||
func TestOriginChangesLoadRecordsNothing(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
table := routetable.New(routetable.Params{
|
|
||||||
PeerNumbers: map[string]uint{peerA: 0},
|
|
||||||
})
|
|
||||||
prefix := netip.MustParsePrefix(prefixV4)
|
|
||||||
|
|
||||||
peers := new(bitset.BitSet)
|
|
||||||
peers.Set(0)
|
|
||||||
|
|
||||||
table.Load(routetable.Change{
|
|
||||||
Prefix: prefix,
|
|
||||||
OriginASN: asA,
|
|
||||||
Peers: peers,
|
|
||||||
Routes: []routetable.Route{routeOf(peerA, asA, 10)},
|
|
||||||
})
|
|
||||||
|
|
||||||
changes, dropped := table.TakeOriginChanges(math.MaxInt)
|
|
||||||
if dropped != 0 || len(changes) != 0 {
|
|
||||||
t.Errorf("TakeOriginChanges = %+v, %d; want none", changes, dropped)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOriginChangesTakeReturnsOldestFirst checks that TakeOriginChanges
|
|
||||||
// returns the changes oldest first and at most max.
|
|
||||||
func TestOriginChangesTakeReturnsOldestFirst(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
table := routetable.New(routetable.Params{})
|
|
||||||
prefix := netip.MustParsePrefix(prefixV4)
|
|
||||||
|
|
||||||
table.Announce(prefix, asA, routeOf(peerA, asA, 10))
|
|
||||||
table.Announce(prefix, asB, routeOf(peerB, asB, 20))
|
|
||||||
table.Announce(prefix, asC, routeOf(peerC, asC, 30))
|
|
||||||
|
|
||||||
first, dropped := table.TakeOriginChanges(2)
|
|
||||||
if dropped != 0 {
|
|
||||||
t.Errorf("dropped = %d, want 0", dropped)
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(first) != 2 || first[0].OriginASN != asA || first[1].OriginASN != asB {
|
|
||||||
t.Errorf("first TakeOriginChanges = %+v, want the first two", first)
|
|
||||||
}
|
|
||||||
|
|
||||||
second, _ := table.TakeOriginChanges(math.MaxInt)
|
|
||||||
if len(second) != 1 || second[0].OriginASN != asC {
|
|
||||||
t.Errorf("second TakeOriginChanges = %+v, want the third", second)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOriginChangesPutBackPutsRowsInFront checks that PutBackOriginChanges
|
|
||||||
// puts rows back in front, so the next TakeOriginChanges returns them first.
|
|
||||||
func TestOriginChangesPutBackPutsRowsInFront(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
table := routetable.New(routetable.Params{})
|
|
||||||
prefix := netip.MustParsePrefix(prefixV4)
|
|
||||||
|
|
||||||
table.Announce(prefix, asA, routeOf(peerA, asA, 10))
|
|
||||||
table.Announce(prefix, asB, routeOf(peerB, asB, 20))
|
|
||||||
|
|
||||||
taken, _ := table.TakeOriginChanges(math.MaxInt)
|
|
||||||
table.Announce(prefix, asC, routeOf(peerC, asC, 30))
|
|
||||||
table.PutBackOriginChanges(taken)
|
|
||||||
|
|
||||||
changes, _ := table.TakeOriginChanges(math.MaxInt)
|
|
||||||
if len(changes) != 3 || changes[0].OriginASN != asA ||
|
|
||||||
changes[1].OriginASN != asB || changes[2].OriginASN != asC {
|
|
||||||
t.Errorf("TakeOriginChanges = %+v, want the put-back rows first", changes)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOriginChangesDropPastTheLimit checks that a change past 100,000 waiting
|
|
||||||
// rows is dropped and counted.
|
|
||||||
func TestOriginChangesDropPastTheLimit(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
table := routetable.New(routetable.Params{})
|
|
||||||
addr := netip.MustParseAddr("10.0.0.0")
|
|
||||||
|
|
||||||
for range routetable.MaxWaitingOriginChanges + 1 {
|
|
||||||
prefix := netip.PrefixFrom(addr, 32)
|
|
||||||
table.Announce(prefix, asA, routeOf(peerA, asA, 10))
|
|
||||||
|
|
||||||
addr = addr.Next()
|
|
||||||
}
|
|
||||||
|
|
||||||
changes, dropped := table.TakeOriginChanges(routetable.MaxWaitingOriginChanges)
|
|
||||||
if dropped != 1 {
|
|
||||||
t.Errorf("dropped = %d, want 1", dropped)
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(changes) != routetable.MaxWaitingOriginChanges {
|
|
||||||
t.Errorf("len(changes) = %d, want %d", len(changes),
|
|
||||||
routetable.MaxWaitingOriginChanges)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOriginChangesTakeResetsTheDroppedCount checks that a TakeOriginChanges
|
|
||||||
// that reports dropped rows leaves the next one reporting none, so each drop
|
|
||||||
// is logged once, not on every later write.
|
|
||||||
func TestOriginChangesTakeResetsTheDroppedCount(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
table := routetable.New(routetable.Params{})
|
|
||||||
addr := netip.MustParseAddr("10.0.0.0")
|
|
||||||
|
|
||||||
for range routetable.MaxWaitingOriginChanges + 1 {
|
|
||||||
table.Announce(netip.PrefixFrom(addr, 32), asA, routeOf(peerA, asA, 10))
|
|
||||||
addr = addr.Next()
|
|
||||||
}
|
|
||||||
|
|
||||||
_, dropped := table.TakeOriginChanges(routetable.MaxWaitingOriginChanges)
|
|
||||||
if dropped != 1 {
|
|
||||||
t.Fatalf("dropped = %d, want 1", dropped)
|
|
||||||
}
|
|
||||||
|
|
||||||
_, dropped = table.TakeOriginChanges(0)
|
|
||||||
if dropped != 0 {
|
|
||||||
t.Errorf("the next take's dropped = %d, want 0", dropped)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOriginChangesPutBackTrimsToTheLimit checks that putting back what a
|
|
||||||
// failed write took, after new changes filled the table again, drops the
|
|
||||||
// oldest changes past the cap and counts them, so the cap still holds. The
|
|
||||||
// clock advances one second per change, so a change's time says its age.
|
|
||||||
func TestOriginChangesPutBackTrimsToTheLimit(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
const (
|
|
||||||
taken = 20000
|
|
||||||
added = 20000
|
|
||||||
)
|
|
||||||
|
|
||||||
now := time.Unix(0, 0)
|
|
||||||
table := routetable.New(routetable.Params{
|
|
||||||
Now: func() time.Time {
|
|
||||||
now = now.Add(time.Second)
|
|
||||||
|
|
||||||
return now
|
|
||||||
},
|
|
||||||
})
|
|
||||||
addr := netip.MustParseAddr("10.0.0.0")
|
|
||||||
|
|
||||||
announce := func(n int) {
|
|
||||||
for range n {
|
|
||||||
table.Announce(netip.PrefixFrom(addr, 32), asA, routeOf(peerA, asA, 10))
|
|
||||||
addr = addr.Next()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
announce(routetable.MaxWaitingOriginChanges)
|
|
||||||
|
|
||||||
takenChanges, _ := table.TakeOriginChanges(taken)
|
|
||||||
|
|
||||||
announce(added)
|
|
||||||
|
|
||||||
table.PutBackOriginChanges(takenChanges)
|
|
||||||
|
|
||||||
changes, dropped := table.TakeOriginChanges(math.MaxInt)
|
|
||||||
if dropped != added {
|
|
||||||
t.Errorf("dropped = %d, want %d", dropped, added)
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(changes) != routetable.MaxWaitingOriginChanges {
|
|
||||||
t.Errorf("len(changes) = %d, want %d", len(changes),
|
|
||||||
routetable.MaxWaitingOriginChanges)
|
|
||||||
}
|
|
||||||
|
|
||||||
// The put-back rows are the oldest, so they are the ones dropped. The
|
|
||||||
// first remaining change is the one that followed them.
|
|
||||||
if want := time.Unix(int64(added+1), 0); !changes[0].Time.Equal(want) {
|
|
||||||
t.Errorf("first change's time = %v, want %v", changes[0].Time, want)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// checkOriginChanges checks that TakeOriginChanges, taking every change,
|
|
||||||
// returns want.
|
|
||||||
func checkOriginChanges(
|
|
||||||
t *testing.T, table *routetable.Table, want []routetable.OriginChange,
|
|
||||||
) {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
changes, dropped := table.TakeOriginChanges(math.MaxInt)
|
|
||||||
if dropped != 0 {
|
|
||||||
t.Errorf("dropped = %d, want 0", dropped)
|
|
||||||
}
|
|
||||||
|
|
||||||
if !reflect.DeepEqual(changes, want) {
|
|
||||||
t.Errorf("TakeOriginChanges = %+v, want %+v", changes, want)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -2,9 +2,7 @@
|
|||||||
// and origin AS, it keeps the peers whose current route to the prefix has that
|
// and origin AS, it keeps the peers whose current route to the prefix has that
|
||||||
// origin AS. It also keeps the two newest of those routes in full. It applies
|
// origin AS. It also keeps the two newest of those routes in full. It applies
|
||||||
// announcements, withdrawals and peers' sessions going down. It answers the
|
// announcements, withdrawals and peers' sessions going down. It answers the
|
||||||
// route reads that the pages and the API need. It also records each prefix and
|
// route reads that the pages and the API need.
|
||||||
// origin AS pair's origin changes: when it gains its first peer and when it
|
|
||||||
// loses its last.
|
|
||||||
package routetable
|
package routetable
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -93,9 +91,6 @@ type Params struct {
|
|||||||
// peer not in it gets the next number above all of them the first time it
|
// peer not in it gets the next number above all of them the first time it
|
||||||
// announces a route.
|
// announces a route.
|
||||||
PeerNumbers map[string]uint
|
PeerNumbers map[string]uint
|
||||||
// Now returns the time RouteWatch applies a change at. It is time.Now
|
|
||||||
// unless a test gives one.
|
|
||||||
Now func() time.Time
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Table is the routing table. Its methods are safe for concurrent use.
|
// Table is the routing table. Its methods are safe for concurrent use.
|
||||||
@@ -131,24 +126,11 @@ type Table struct {
|
|||||||
// originsWithoutKeptRoute counts the origins that peers carry but that
|
// originsWithoutKeptRoute counts the origins that peers carry but that
|
||||||
// keep no route.
|
// keep no route.
|
||||||
originsWithoutKeptRoute int
|
originsWithoutKeptRoute int
|
||||||
|
|
||||||
// now returns the time RouteWatch applies a change at.
|
|
||||||
now func() time.Time
|
|
||||||
// originChanges holds the origin changes that wait to be written, oldest
|
|
||||||
// first. originChangesDropped counts those dropped since the last
|
|
||||||
// TakeOriginChanges.
|
|
||||||
originChanges []OriginChange
|
|
||||||
originChangesDropped int
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// New returns an empty table that numbers new peers on from
|
// New returns an empty table that numbers new peers on from
|
||||||
// params.PeerNumbers.
|
// params.PeerNumbers.
|
||||||
func New(params Params) *Table {
|
func New(params Params) *Table {
|
||||||
now := params.Now
|
|
||||||
if now == nil {
|
|
||||||
now = time.Now
|
|
||||||
}
|
|
||||||
|
|
||||||
t := &Table{
|
t := &Table{
|
||||||
ipv4: newFamily(ipv4Bits, minLookupMaskLengthV4),
|
ipv4: newFamily(ipv4Bits, minLookupMaskLengthV4),
|
||||||
ipv6: newFamily(ipv6Bits, minLookupMaskLengthV6),
|
ipv6: newFamily(ipv6Bits, minLookupMaskLengthV6),
|
||||||
@@ -157,7 +139,6 @@ func New(params Params) *Table {
|
|||||||
peerAddresses: make(map[uint]string, len(params.PeerNumbers)),
|
peerAddresses: make(map[uint]string, len(params.PeerNumbers)),
|
||||||
newPeers: make(map[string]uint),
|
newPeers: make(map[string]uint),
|
||||||
changed: make(map[pair]struct{}),
|
changed: make(map[pair]struct{}),
|
||||||
now: now,
|
|
||||||
keptTimes: btree.NewG(keptTimesDegree, func(a, b keptTime) bool {
|
keptTimes: btree.NewG(keptTimesDegree, func(a, b keptTime) bool {
|
||||||
return a.time.Before(b.time)
|
return a.time.Before(b.time)
|
||||||
}),
|
}),
|
||||||
@@ -287,9 +268,8 @@ func (t *Table) PeersDown(peers []string) {
|
|||||||
// Load puts one prefix and origin AS pair that the database holds into the
|
// Load puts one prefix and origin AS pair that the database holds into the
|
||||||
// table, as startup does with each row it reads. The change's peers join the
|
// table, as startup does with each row it reads. The change's peers join the
|
||||||
// pair's peer set, and its routes are kept as an announcement's would be.
|
// pair's peer set, and its routes are kept as an announcement's would be.
|
||||||
// Unlike an update, Load records no change and no origin change, so
|
// Unlike an update, Load records no change, so TakeChanges does not hand the
|
||||||
// TakeChanges does not hand the pair back to be written again. Loading a pair
|
// pair back to be written again. Loading a pair twice changes nothing.
|
||||||
// twice changes nothing.
|
|
||||||
//
|
//
|
||||||
// The caller makes sure that every peer change names, in its peer set or in
|
// The caller makes sure that every peer change names, in its peer set or in
|
||||||
// its routes, has a number in Params.PeerNumbers.
|
// its routes, has a number in Params.PeerNumbers.
|
||||||
@@ -480,7 +460,6 @@ func (t *Table) removeOrigin(prefix netip.Prefix, o *origin) {
|
|||||||
// r in full when it is one of the two newest routes of o. The peer's earlier
|
// r in full when it is one of the two newest routes of o. The peer's earlier
|
||||||
// route, if o keeps one, is replaced: it is no longer the peer's current route.
|
// route, if o keeps one, is replaced: it is no longer the peer's current route.
|
||||||
// The pair is recorded as changed only when its peer set or kept routes change.
|
// The pair is recorded as changed only when its peer set or kept routes change.
|
||||||
// When r is the origin's first peer, an announced origin change is recorded.
|
|
||||||
func (t *Table) join(prefix netip.Prefix, o *origin, r route) {
|
func (t *Table) join(prefix netip.Prefix, o *origin, r route) {
|
||||||
if slices.ContainsFunc(o.routes, r.equal) {
|
if slices.ContainsFunc(o.routes, r.equal) {
|
||||||
// The peer announced its kept route again.
|
// The peer announced its kept route again.
|
||||||
@@ -490,8 +469,6 @@ func (t *Table) join(prefix netip.Prefix, o *origin, r route) {
|
|||||||
wasWithoutKeptRoute := o.withoutKeptRoute()
|
wasWithoutKeptRoute := o.withoutKeptRoute()
|
||||||
|
|
||||||
joined := !o.peers.Test(r.peer)
|
joined := !o.peers.Test(r.peer)
|
||||||
|
|
||||||
firstPeer := joined && o.peers.None()
|
|
||||||
if joined {
|
if joined {
|
||||||
o.peers.Set(r.peer)
|
o.peers.Set(r.peer)
|
||||||
|
|
||||||
@@ -505,10 +482,6 @@ func (t *Table) join(prefix netip.Prefix, o *origin, r route) {
|
|||||||
if joined || dropped || kept {
|
if joined || dropped || kept {
|
||||||
t.recordChange(prefix, o, wasWithoutKeptRoute)
|
t.recordChange(prefix, o, wasWithoutKeptRoute)
|
||||||
}
|
}
|
||||||
|
|
||||||
if firstPeer {
|
|
||||||
t.recordOriginChange(prefix, o.asn, true, t.prefixPeerCount(prefix))
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// leave takes peer out of the peer set of o, an origin of prefix, and drops
|
// leave takes peer out of the peer set of o, an origin of prefix, and drops
|
||||||
@@ -567,8 +540,7 @@ func (t *Table) dropKeptRoute(o *origin, peer uint) bool {
|
|||||||
// recordChange records that the pair of prefix and the AS of o changed, and
|
// recordChange records that the pair of prefix and the AS of o changed, and
|
||||||
// brings the count of origins without a kept route up to date.
|
// brings the count of origins without a kept route up to date.
|
||||||
// wasWithoutKeptRoute says whether o was such an origin before the change. An
|
// wasWithoutKeptRoute says whether o was such an origin before the change. An
|
||||||
// origin that no peer carries any more is removed, and a withdrawn origin
|
// origin that no peer carries any more is removed.
|
||||||
// change is recorded.
|
|
||||||
func (t *Table) recordChange(
|
func (t *Table) recordChange(
|
||||||
prefix netip.Prefix, o *origin, wasWithoutKeptRoute bool,
|
prefix netip.Prefix, o *origin, wasWithoutKeptRoute bool,
|
||||||
) {
|
) {
|
||||||
@@ -577,7 +549,6 @@ func (t *Table) recordChange(
|
|||||||
t.countOriginWithoutKeptRoute(o, wasWithoutKeptRoute)
|
t.countOriginWithoutKeptRoute(o, wasWithoutKeptRoute)
|
||||||
|
|
||||||
if o.peers.None() {
|
if o.peers.None() {
|
||||||
t.recordOriginChange(prefix, o.asn, false, t.prefixPeerCount(prefix))
|
|
||||||
t.removeOrigin(prefix, o)
|
t.removeOrigin(prefix, o)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -330,23 +330,6 @@ func (m *mockStore) ReadRouteTable(_ context.Context) (*routetable.Table, error)
|
|||||||
return routetable.New(routetable.Params{}), nil
|
return routetable.New(routetable.Params{}), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// WriteOriginChanges records nothing: the mock keeps no origin changes.
|
|
||||||
func (m *mockStore) WriteOriginChanges(_ []routetable.OriginChange) error {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetOriginChanges returns no origin changes.
|
|
||||||
func (m *mockStore) GetOriginChanges(
|
|
||||||
_ context.Context, _ netip.Prefix, _ int,
|
|
||||||
) ([]routetable.OriginChange, error) {
|
|
||||||
return nil, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// PruneOriginChanges does nothing.
|
|
||||||
func (m *mockStore) PruneOriginChanges(_ context.Context) error {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// GetOrCreateASNBatch mock implementation
|
// GetOrCreateASNBatch mock implementation
|
||||||
func (m *mockStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error {
|
func (m *mockStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error {
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
|
|||||||
@@ -175,13 +175,12 @@ func (s *blockingStore) releaseWrites() {
|
|||||||
//
|
//
|
||||||
//nolint:tagliatelle // field names are the keys the handlers log
|
//nolint:tagliatelle // field names are the keys the handlers log
|
||||||
type stopLog struct {
|
type stopLog struct {
|
||||||
Msg string `json:"msg"`
|
Msg string `json:"msg"`
|
||||||
Updates int `json:"updates"`
|
Updates int `json:"updates"`
|
||||||
TableChanges int `json:"table_changes"`
|
TableChanges int `json:"table_changes"`
|
||||||
HeldMessages int `json:"held_messages"`
|
HeldMessages int `json:"held_messages"`
|
||||||
OriginChanges int `json:"origin_changes"`
|
Paths int `json:"paths"`
|
||||||
Paths int `json:"paths"`
|
BatchEntries int `json:"batch_entries"`
|
||||||
BatchEntries int `json:"batch_entries"`
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// newErrorLogger returns a logger that writes only errors, as JSON lines, to
|
// newErrorLogger returns a logger that writes only errors, as JSON lines, to
|
||||||
@@ -473,42 +472,6 @@ func (s tableWriteBlockingStore) WriteRouteChanges(
|
|||||||
return s.mockStore.WriteRouteChanges(changes, newPeers)
|
return s.mockStore.WriteRouteChanges(changes, newPeers)
|
||||||
}
|
}
|
||||||
|
|
||||||
// stopPrefixHandlerWithBlockedWrite stops the prefix handler, whose store
|
|
||||||
// holds the write that fill starts, after handing it the message extra, and
|
|
||||||
// returns what Stop logged. blocking is the store the write waits on; store is
|
|
||||||
// a blockingStore wrapped for the routing table's changes or the origin
|
|
||||||
// changes.
|
|
||||||
func stopPrefixHandlerWithBlockedWrite(
|
|
||||||
t *testing.T,
|
|
||||||
blocking *blockingStore,
|
|
||||||
store database.Store,
|
|
||||||
logs *bytes.Buffer,
|
|
||||||
fill, extra *ristypes.RISMessage,
|
|
||||||
) stopLog {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
h := NewPrefixHandler(store, newErrorLogger(logs),
|
|
||||||
routetable.New(routetable.Params{}))
|
|
||||||
stopWhileAWriteIsBlocked(t, blocking, h, h.done,
|
|
||||||
[]*ristypes.RISMessage{fill}, extra)
|
|
||||||
|
|
||||||
return lastLogged(t, logs)
|
|
||||||
}
|
|
||||||
|
|
||||||
// originChangesBlockingStore is a blockingStore whose writes of the origin
|
|
||||||
// changes also wait until the test calls releaseWrites.
|
|
||||||
type originChangesBlockingStore struct {
|
|
||||||
*blockingStore
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s originChangesBlockingStore) WriteOriginChanges(
|
|
||||||
changes []routetable.OriginChange,
|
|
||||||
) error {
|
|
||||||
s.waitForRelease()
|
|
||||||
|
|
||||||
return s.mockStore.WriteOriginChanges(changes)
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress holds the write
|
// TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress holds the write
|
||||||
// of the routing table's changes that a full batch starts, then hands the
|
// of the routing table's changes that a full batch starts, then hands the
|
||||||
// prefix handler one more prefix and stops it with a deadline that passes
|
// prefix handler one more prefix and stops it with a deadline that passes
|
||||||
@@ -522,67 +485,19 @@ func TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress(t *testing.T)
|
|||||||
var logs bytes.Buffer
|
var logs bytes.Buffer
|
||||||
|
|
||||||
blocking := newBlockingStore()
|
blocking := newBlockingStore()
|
||||||
|
h := NewPrefixHandler(tableWriteBlockingStore{blockingStore: blocking},
|
||||||
|
newErrorLogger(&logs), routetable.New(routetable.Params{}))
|
||||||
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
|
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
|
||||||
logged := stopPrefixHandlerWithBlockedWrite(t, blocking,
|
stopWhileAWriteIsBlocked(t, blocking, h, h.done,
|
||||||
tableWriteBlockingStore{blockingStore: blocking}, &logs,
|
[]*ristypes.RISMessage{prefixAnnouncement(fill)},
|
||||||
prefixAnnouncement(fill), prefixAnnouncement([]string{extraPrefix}))
|
prefixAnnouncement([]string{extraPrefix}))
|
||||||
|
|
||||||
if logged.TableChanges != 2 {
|
if logged := lastLogged(t, &logs); logged.TableChanges != 2 {
|
||||||
t.Errorf("Stop logged %d table changes not written, want 2",
|
t.Errorf("Stop logged %d table changes not written, want 2",
|
||||||
logged.TableChanges)
|
logged.TableChanges)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestPrefixHandlerStopCountsTheOriginChangesOfAWriteInProgress holds the
|
|
||||||
// write of the origin changes that a full batch starts, then hands the prefix
|
|
||||||
// handler one more prefix and stops it with a deadline that passes meanwhile.
|
|
||||||
// Stop must log as not written both the change the held write took and the one
|
|
||||||
// since.
|
|
||||||
func TestPrefixHandlerStopCountsTheOriginChangesOfAWriteInProgress(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
const extraPrefix = "198.51.100.0/24"
|
|
||||||
|
|
||||||
var logs bytes.Buffer
|
|
||||||
|
|
||||||
blocking := newBlockingStore()
|
|
||||||
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
|
|
||||||
logged := stopPrefixHandlerWithBlockedWrite(t, blocking,
|
|
||||||
originChangesBlockingStore{blockingStore: blocking}, &logs,
|
|
||||||
prefixAnnouncement(fill), prefixAnnouncement([]string{extraPrefix}))
|
|
||||||
|
|
||||||
if logged.OriginChanges != 2 {
|
|
||||||
t.Errorf("Stop logged %d origin changes not written, want 2",
|
|
||||||
logged.OriginChanges)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixHandlerStopCountsTheWaitingOriginChanges stops the prefix handler
|
|
||||||
// with a deadline that has already passed, so the flush goroutine skips its
|
|
||||||
// last write. The origin changes still waiting in the routing table are then
|
|
||||||
// not written, and Stop must log them.
|
|
||||||
func TestPrefixHandlerStopCountsTheWaitingOriginChanges(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
var logs bytes.Buffer
|
|
||||||
|
|
||||||
store := newMockStore()
|
|
||||||
h := newPrefixHandler(store, newErrorLogger(&logs),
|
|
||||||
routetable.New(routetable.Params{}), longerThanAnyTest,
|
|
||||||
maxTableChangesPerWrite)
|
|
||||||
|
|
||||||
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix, testV6Prefix))
|
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(t.Context())
|
|
||||||
cancel()
|
|
||||||
h.Stop(ctx)
|
|
||||||
|
|
||||||
if logged := lastLogged(t, &logs); logged.OriginChanges != 2 {
|
|
||||||
t.Errorf("Stop logged %d origin changes not written, want 2",
|
|
||||||
logged.OriginChanges)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestASHandlerSkipsItsLastWriteAfterTheStopDeadline also checks what Stop
|
// TestASHandlerSkipsItsLastWriteAfterTheStopDeadline also checks what Stop
|
||||||
// logs: the asnBatchSize entries of the held write and the entry for testASNB,
|
// logs: the asnBatchSize entries of the held write and the entry for testASNB,
|
||||||
// handled since, are not written.
|
// handled since, are not written.
|
||||||
|
|||||||
@@ -25,10 +25,6 @@ const (
|
|||||||
// analyzeInterval is how often to run ANALYZE.
|
// analyzeInterval is how often to run ANALYZE.
|
||||||
analyzeInterval = 1 * time.Hour
|
analyzeInterval = 1 * time.Hour
|
||||||
|
|
||||||
// originChangesPruneInterval is how often to prune the origin_changes
|
|
||||||
// table.
|
|
||||||
originChangesPruneInterval = 1 * time.Hour
|
|
||||||
|
|
||||||
// checkpointTimeout is the max time for WAL checkpoint.
|
// checkpointTimeout is the max time for WAL checkpoint.
|
||||||
checkpointTimeout = 10 * time.Second
|
checkpointTimeout = 10 * time.Second
|
||||||
|
|
||||||
@@ -37,9 +33,6 @@ const (
|
|||||||
|
|
||||||
// analyzeTimeout is the max time for ANALYZE.
|
// analyzeTimeout is the max time for ANALYZE.
|
||||||
analyzeTimeout = 5 * time.Minute
|
analyzeTimeout = 5 * time.Minute
|
||||||
|
|
||||||
// originChangesPruneTimeout is the max time for the origin changes prune.
|
|
||||||
originChangesPruneTimeout = 5 * time.Minute
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// checkpointIncompleteMessage is logged, at debug level like a completed
|
// checkpointIncompleteMessage is logged, at debug level like a completed
|
||||||
@@ -87,7 +80,6 @@ func (m *DBMaintainer) Start(ctx context.Context) {
|
|||||||
"checkpoint_interval", checkpointInterval,
|
"checkpoint_interval", checkpointInterval,
|
||||||
"vacuum_interval", vacuumInterval,
|
"vacuum_interval", vacuumInterval,
|
||||||
"analyze_interval", analyzeInterval,
|
"analyze_interval", analyzeInterval,
|
||||||
"origin_changes_prune_interval", originChangesPruneInterval,
|
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -119,9 +111,6 @@ func (m *DBMaintainer) run(ctx context.Context) {
|
|||||||
analyzeTimer := time.NewTimer(analyzeInterval)
|
analyzeTimer := time.NewTimer(analyzeInterval)
|
||||||
defer analyzeTimer.Stop()
|
defer analyzeTimer.Stop()
|
||||||
|
|
||||||
pruneTimer := time.NewTimer(originChangesPruneInterval)
|
|
||||||
defer pruneTimer.Stop()
|
|
||||||
|
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-m.stopCh:
|
case <-m.stopCh:
|
||||||
@@ -138,10 +127,6 @@ func (m *DBMaintainer) run(ctx context.Context) {
|
|||||||
case <-analyzeTimer.C:
|
case <-analyzeTimer.C:
|
||||||
m.runAnalyze(ctx)
|
m.runAnalyze(ctx)
|
||||||
analyzeTimer.Reset(analyzeInterval)
|
analyzeTimer.Reset(analyzeInterval)
|
||||||
|
|
||||||
case <-pruneTimer.C:
|
|
||||||
m.runPruneOriginChanges(ctx)
|
|
||||||
pruneTimer.Reset(originChangesPruneInterval)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -190,18 +175,6 @@ func (m *DBMaintainer) runAnalyze(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// runPruneOriginChanges prunes the origin_changes table.
|
|
||||||
func (m *DBMaintainer) runPruneOriginChanges(ctx context.Context) {
|
|
||||||
duration, err := runWithTimeout(ctx, originChangesPruneTimeout,
|
|
||||||
m.db.PruneOriginChanges)
|
|
||||||
if err != nil {
|
|
||||||
m.logger.Error("Origin changes prune failed", "error", err,
|
|
||||||
"duration", duration)
|
|
||||||
} else {
|
|
||||||
m.logger.Debug("Origin changes pruned", "duration", duration)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// runWithTimeout runs operation with a context that ends after timeout and
|
// runWithTimeout runs operation with a context that ends after timeout and
|
||||||
// returns how long it took and its error.
|
// returns how long it took and its error.
|
||||||
func runWithTimeout(
|
func runWithTimeout(
|
||||||
|
|||||||
@@ -48,10 +48,6 @@ const (
|
|||||||
// going down changes are written in about a minute.
|
// going down changes are written in about a minute.
|
||||||
maxTableChangesPerWrite = 20000
|
maxTableChangesPerWrite = 20000
|
||||||
|
|
||||||
// maxOriginChangesPerWrite is the most origin changes of the routing
|
|
||||||
// table that one flush writes; the rest wait for the next flushes.
|
|
||||||
maxOriginChangesPerWrite = 20000
|
|
||||||
|
|
||||||
// maxHeldBackMessages is the most messages the handler holds back from
|
// maxHeldBackMessages is the most messages the handler holds back from
|
||||||
// peers whose session went down. A peer's messages are held back from
|
// peers whose session went down. A peer's messages are held back from
|
||||||
// the moment its session goes down until peerDown has applied them. If
|
// the moment its session goes down until peerDown has applied them. If
|
||||||
@@ -85,8 +81,8 @@ const (
|
|||||||
// a time, and takes new messages in between. It holds back at most
|
// a time, and takes new messages in between. It holds back at most
|
||||||
// maxHeldBackMessages messages.
|
// maxHeldBackMessages messages.
|
||||||
//
|
//
|
||||||
// Each flush writes the table's changes and origin changes to the database.
|
// Each flush writes the table's changes to the database. It also stores each
|
||||||
// It also stores each prefix the updates name with the time it was first seen.
|
// prefix the updates name with the time it was first seen.
|
||||||
type PrefixHandler struct {
|
type PrefixHandler struct {
|
||||||
db database.Store
|
db database.Store
|
||||||
logger *logger.Logger
|
logger *logger.Logger
|
||||||
@@ -99,14 +95,11 @@ type PrefixHandler struct {
|
|||||||
// tableChangesPerWrite is maxTableChangesPerWrite, except in tests that
|
// tableChangesPerWrite is maxTableChangesPerWrite, except in tests that
|
||||||
// give a smaller one.
|
// give a smaller one.
|
||||||
tableChangesPerWrite int
|
tableChangesPerWrite int
|
||||||
// originChangesPerWrite is maxOriginChangesPerWrite, except in tests that
|
|
||||||
// give a smaller one.
|
|
||||||
originChangesPerWrite int
|
|
||||||
|
|
||||||
// Batching. mu guards batch, writing, writingTableChanges and
|
// Batching. mu guards batch, writing and writingTableChanges: the
|
||||||
// writingOriginChanges: the flush goroutine takes them under mu and
|
// flush goroutine takes them under mu and writes them after releasing
|
||||||
// writes them after releasing mu, so HandleMessage waits for a database
|
// mu, so HandleMessage waits for a database write only while batch is at
|
||||||
// write only while batch is at batchMaxSize.
|
// batchMaxSize.
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
batch []seenPrefix
|
batch []seenPrefix
|
||||||
// writing is how many updates the flush goroutine took and has not
|
// writing is how many updates the flush goroutine took and has not
|
||||||
@@ -115,9 +108,6 @@ type PrefixHandler struct {
|
|||||||
// writingTableChanges is how many changes the flush goroutine took from
|
// writingTableChanges is how many changes the flush goroutine took from
|
||||||
// the routing table and has not finished writing.
|
// the routing table and has not finished writing.
|
||||||
writingTableChanges int
|
writingTableChanges int
|
||||||
// writingOriginChanges is how many origin changes the flush goroutine
|
|
||||||
// took from the routing table and has not finished writing.
|
|
||||||
writingOriginChanges int
|
|
||||||
// batchTaken is signalled when the flush goroutine takes batch; its L is
|
// batchTaken is signalled when the flush goroutine takes batch; its L is
|
||||||
// &mu.
|
// &mu.
|
||||||
batchTaken sync.Cond
|
batchTaken sync.Cond
|
||||||
@@ -200,18 +190,17 @@ func newPrefixHandler(
|
|||||||
batchTimeout time.Duration, tableChangesPerWrite int,
|
batchTimeout time.Duration, tableChangesPerWrite int,
|
||||||
) *PrefixHandler {
|
) *PrefixHandler {
|
||||||
h := &PrefixHandler{
|
h := &PrefixHandler{
|
||||||
db: db,
|
db: db,
|
||||||
logger: logger,
|
logger: logger,
|
||||||
table: table,
|
table: table,
|
||||||
batch: make([]seenPrefix, 0, prefixBatchSize),
|
batch: make([]seenPrefix, 0, prefixBatchSize),
|
||||||
batchMaxSize: prefixBatchMaxSize,
|
batchMaxSize: prefixBatchMaxSize,
|
||||||
tableChangesPerWrite: tableChangesPerWrite,
|
tableChangesPerWrite: tableChangesPerWrite,
|
||||||
originChangesPerWrite: maxOriginChangesPerWrite,
|
batchFullCh: make(chan struct{}, 1),
|
||||||
batchFullCh: make(chan struct{}, 1),
|
stopCh: make(chan context.Context, 1),
|
||||||
stopCh: make(chan context.Context, 1),
|
done: make(chan struct{}),
|
||||||
done: make(chan struct{}),
|
heldBack: make(map[string][]*ristypes.RISMessage),
|
||||||
heldBack: make(map[string][]*ristypes.RISMessage),
|
tablePeersDown: table.PeersDown,
|
||||||
tablePeersDown: table.PeersDown,
|
|
||||||
}
|
}
|
||||||
h.batchTaken.L = &h.mu
|
h.batchTaken.L = &h.mu
|
||||||
h.heldTaken.L = &h.mu
|
h.heldTaken.L = &h.mu
|
||||||
@@ -279,13 +268,13 @@ func (h *PrefixHandler) HeldBackCount() int {
|
|||||||
// back meanwhile to be applied. It gives up this wait once ctx ends. Then Stop
|
// back meanwhile to be applied. It gives up this wait once ctx ends. Then Stop
|
||||||
// stops the flush goroutine and waits for it until ctx ends. The flush
|
// stops the flush goroutine and waits for it until ctx ends. The flush
|
||||||
// goroutine makes one last flush before it returns. That flush writes at most
|
// goroutine makes one last flush before it returns. That flush writes at most
|
||||||
// tableChangesPerWrite of the routing table's changes, originChangesPerWrite of
|
// tableChangesPerWrite of the routing table's changes, and the prefixes of the
|
||||||
// its origin changes, and the prefixes of the updates the handler holds. Once
|
// updates the handler holds. Once ctx has ended the flush goroutine skips that
|
||||||
// ctx has ended the flush goroutine skips that last flush, and a write still
|
// last flush, and a write still running is lost when the process exits. A
|
||||||
// running is lost when the process exits. A failed write of the table's
|
// failed write of the table's changes is not retried either. Stop logs how
|
||||||
// changes is not retried either. Stop logs how many updates, table changes and
|
// many updates and table changes are not written. It also logs how many
|
||||||
// origin changes are not written. It also logs how many held-back messages are
|
// held-back messages are not applied. The caller calls Stop once its last
|
||||||
// not applied. The caller calls Stop once its last HandleMessage has returned.
|
// HandleMessage has returned.
|
||||||
func (h *PrefixHandler) Stop(ctx context.Context) {
|
func (h *PrefixHandler) Stop(ctx context.Context) {
|
||||||
peerDownsDone := make(chan struct{})
|
peerDownsDone := make(chan struct{})
|
||||||
|
|
||||||
@@ -310,25 +299,23 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
|
|||||||
unwritten := h.writing + len(h.batch)
|
unwritten := h.writing + len(h.batch)
|
||||||
tableChanges := h.writingTableChanges + h.table.ChangeCount()
|
tableChanges := h.writingTableChanges + h.table.ChangeCount()
|
||||||
heldMessages := h.heldCount
|
heldMessages := h.heldCount
|
||||||
originChanges := h.writingOriginChanges + h.table.OriginChangeCount()
|
|
||||||
h.mu.Unlock()
|
h.mu.Unlock()
|
||||||
|
|
||||||
if unwritten == 0 && tableChanges == 0 && heldMessages == 0 &&
|
if unwritten == 0 && tableChanges == 0 && heldMessages == 0 {
|
||||||
originChanges == 0 {
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if ctx.Err() != nil {
|
if ctx.Err() != nil {
|
||||||
h.logger.Error("Stop deadline passed before the prefix handler wrote "+
|
h.logger.Error("Stop deadline passed before the prefix handler wrote "+
|
||||||
"its updates", "updates", unwritten, "table_changes", tableChanges,
|
"its updates", "updates", unwritten, "table_changes", tableChanges,
|
||||||
"held_messages", heldMessages, "origin_changes", originChanges)
|
"held_messages", heldMessages)
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
h.logger.Error("The prefix handler stopped before it wrote everything",
|
h.logger.Error("The prefix handler stopped before it wrote everything",
|
||||||
"updates", unwritten, "table_changes", tableChanges,
|
"updates", unwritten, "table_changes", tableChanges,
|
||||||
"held_messages", heldMessages, "origin_changes", originChanges)
|
"held_messages", heldMessages)
|
||||||
}
|
}
|
||||||
|
|
||||||
// applyMessage applies msg. The caller holds mu. A peer's session going down
|
// applyMessage applies msg. The caller holds mu. A peer's session going down
|
||||||
@@ -515,12 +502,11 @@ func (h *PrefixHandler) takeBatch() []seenPrefix {
|
|||||||
return batch
|
return batch
|
||||||
}
|
}
|
||||||
|
|
||||||
// flushBatch writes the routing table's changes and origin changes, then
|
// flushBatch writes the routing table's changes, then takes the pending batch
|
||||||
// takes the pending batch and stores the prefixes it names, without holding
|
// and stores the prefixes it names, without holding the mutex during the
|
||||||
// the mutex during the writes.
|
// writes.
|
||||||
func (h *PrefixHandler) flushBatch() {
|
func (h *PrefixHandler) flushBatch() {
|
||||||
h.writeTableChanges()
|
h.writeTableChanges()
|
||||||
h.writeOriginChanges()
|
|
||||||
|
|
||||||
batch := h.takeBatch()
|
batch := h.takeBatch()
|
||||||
if len(batch) == 0 {
|
if len(batch) == 0 {
|
||||||
@@ -622,39 +608,6 @@ func (h *PrefixHandler) writeTableChanges() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// writeOriginChanges writes at most originChangesPerWrite of the routing
|
|
||||||
// table's origin changes. When the write fails, they go back to the table, so
|
|
||||||
// that a later write writes them again. A dropped change is logged as a
|
|
||||||
// warning.
|
|
||||||
func (h *PrefixHandler) writeOriginChanges() {
|
|
||||||
h.mu.Lock()
|
|
||||||
changes, dropped := h.table.TakeOriginChanges(h.originChangesPerWrite)
|
|
||||||
h.writingOriginChanges = len(changes)
|
|
||||||
h.mu.Unlock()
|
|
||||||
|
|
||||||
if dropped > 0 {
|
|
||||||
h.logger.Warn("Origin changes dropped", "dropped", dropped)
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(changes) == 0 {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
err := h.db.WriteOriginChanges(changes)
|
|
||||||
|
|
||||||
h.mu.Lock()
|
|
||||||
defer h.mu.Unlock()
|
|
||||||
|
|
||||||
h.writingOriginChanges = 0
|
|
||||||
|
|
||||||
if err != nil {
|
|
||||||
h.logger.Error("Failed to write the origin changes; the next flush "+
|
|
||||||
"retries", "changes", len(changes), "error", err)
|
|
||||||
|
|
||||||
h.table.PutBackOriginChanges(changes)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// canonicalPrefix returns prefix in the text form net/netip prints, the form
|
// canonicalPrefix returns prefix in the text form net/netip prints, the form
|
||||||
// the prefix tables store and the IP lookup reads them in. The feed sends IPv6
|
// the prefix tables store and the IP lookup reads them in. The feed sends IPv6
|
||||||
// announcements compressed ("2001:db8::/32") but IPv6 withdrawals uncompressed
|
// announcements compressed ("2001:db8::/32") but IPv6 withdrawals uncompressed
|
||||||
|
|||||||
@@ -327,14 +327,13 @@ func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
|
|||||||
table := routetable.New(routetable.Params{})
|
table := routetable.New(routetable.Params{})
|
||||||
|
|
||||||
return &PrefixHandler{
|
return &PrefixHandler{
|
||||||
db: store,
|
db: store,
|
||||||
logger: logger.New(),
|
logger: logger.New(),
|
||||||
table: table,
|
table: table,
|
||||||
batchMaxSize: prefixBatchMaxSize,
|
batchMaxSize: prefixBatchMaxSize,
|
||||||
tableChangesPerWrite: maxTableChangesPerWrite,
|
tableChangesPerWrite: maxTableChangesPerWrite,
|
||||||
originChangesPerWrite: maxOriginChangesPerWrite,
|
heldBack: make(map[string][]*ristypes.RISMessage),
|
||||||
heldBack: make(map[string][]*ristypes.RISMessage),
|
tablePeersDown: table.PeersDown,
|
||||||
tablePeersDown: table.PeersDown,
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,181 +0,0 @@
|
|||||||
package routewatch
|
|
||||||
|
|
||||||
import (
|
|
||||||
"bytes"
|
|
||||||
"encoding/json"
|
|
||||||
"errors"
|
|
||||||
"log/slog"
|
|
||||||
"net/netip"
|
|
||||||
"slices"
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/database"
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/routetable"
|
|
||||||
)
|
|
||||||
|
|
||||||
// TestPrefixHandlerWritesOriginChanges hands a running prefix handler an
|
|
||||||
// announcement and flushes. The flush must write the origin change to the
|
|
||||||
// database.
|
|
||||||
func TestPrefixHandlerWritesOriginChanges(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
db := newTestDatabase(t)
|
|
||||||
h := prefixHandlerWithoutTimer(db)
|
|
||||||
|
|
||||||
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
|
|
||||||
h.flushBatch()
|
|
||||||
|
|
||||||
got, err := db.GetOriginChanges(t.Context(),
|
|
||||||
netip.MustParsePrefix(testV4Prefix), 10)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("GetOriginChanges: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(got) != 1 || !got[0].Announced || got[0].OriginASN != testASNB {
|
|
||||||
t.Errorf("GetOriginChanges = %+v, want one announced row for AS%d",
|
|
||||||
got, testASNB)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// errTestOriginChangeWrite is the error a test store's failing write of the
|
|
||||||
// origin changes returns.
|
|
||||||
var errTestOriginChangeWrite = errors.New(
|
|
||||||
"test: the write of the origin changes failed")
|
|
||||||
|
|
||||||
// originChangesFailsStore is a real database whose first write of the origin
|
|
||||||
// changes fails, writing nothing. writes counts the writes asked for.
|
|
||||||
type originChangesFailsStore struct {
|
|
||||||
*database.Database
|
|
||||||
|
|
||||||
writes int
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *originChangesFailsStore) WriteOriginChanges(
|
|
||||||
changes []routetable.OriginChange,
|
|
||||||
) error {
|
|
||||||
s.writes++
|
|
||||||
if s.writes == 1 {
|
|
||||||
return errTestOriginChangeWrite
|
|
||||||
}
|
|
||||||
|
|
||||||
return s.Database.WriteOriginChanges(changes)
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixHandlerWritesAFailedOriginChangeWriteAgain runs two flushes into a
|
|
||||||
// database whose first write of the origin changes fails. The second write must
|
|
||||||
// hold the change of the first.
|
|
||||||
func TestPrefixHandlerWritesAFailedOriginChangeWriteAgain(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
db := newTestDatabase(t)
|
|
||||||
store := &originChangesFailsStore{Database: db}
|
|
||||||
h := prefixHandlerWithoutTimer(store)
|
|
||||||
|
|
||||||
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
|
|
||||||
h.flushBatch()
|
|
||||||
h.flushBatch()
|
|
||||||
|
|
||||||
got, err := db.GetOriginChanges(t.Context(),
|
|
||||||
netip.MustParsePrefix(testV4Prefix), 10)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("GetOriginChanges: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if len(got) != 1 || !got[0].Announced || got[0].OriginASN != testASNB {
|
|
||||||
t.Errorf("GetOriginChanges = %+v, want one announced row for AS%d",
|
|
||||||
got, testASNB)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// originChangesWriteSizesStore is a real database that records how many
|
|
||||||
// changes each write of the origin changes holds.
|
|
||||||
type originChangesWriteSizesStore struct {
|
|
||||||
*database.Database
|
|
||||||
|
|
||||||
counts []int
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *originChangesWriteSizesStore) WriteOriginChanges(
|
|
||||||
changes []routetable.OriginChange,
|
|
||||||
) error {
|
|
||||||
s.counts = append(s.counts, len(changes))
|
|
||||||
|
|
||||||
return s.Database.WriteOriginChanges(changes)
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixHandlerWritesManyOriginChangesOverSeveralFlushes has more origin
|
|
||||||
// changes than one write takes, then flushes three times. The first write must
|
|
||||||
// hold one write's worth, the second the rest, and the third nothing. The
|
|
||||||
// handler's limit is small.
|
|
||||||
func TestPrefixHandlerWritesManyOriginChangesOverSeveralFlushes(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
const perWrite = 10
|
|
||||||
|
|
||||||
db := newTestDatabase(t)
|
|
||||||
store := &originChangesWriteSizesStore{Database: db}
|
|
||||||
h := prefixHandlerWithoutTimer(store)
|
|
||||||
h.originChangesPerWrite = perWrite
|
|
||||||
|
|
||||||
announceManyPrefixes(h.table, perWrite*3/2)
|
|
||||||
|
|
||||||
h.flushBatch()
|
|
||||||
h.flushBatch()
|
|
||||||
h.flushBatch()
|
|
||||||
|
|
||||||
want := []int{perWrite, perWrite*3/2 - perWrite}
|
|
||||||
if !slices.Equal(store.counts, want) {
|
|
||||||
t.Errorf("WriteOriginChanges held %v changes, want %v", store.counts, want)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixHandlerLogsTheDroppedOriginChanges fills the routing table with
|
|
||||||
// one origin change more than it keeps, then flushes. The warning the flush
|
|
||||||
// logs must carry the dropped count.
|
|
||||||
func TestPrefixHandlerLogsTheDroppedOriginChanges(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
var logs bytes.Buffer
|
|
||||||
|
|
||||||
db := newTestDatabase(t)
|
|
||||||
h := prefixHandlerWithoutTimer(db)
|
|
||||||
h.logger = &logger.Logger{
|
|
||||||
Logger: slog.New(slog.NewJSONHandler(&logs,
|
|
||||||
&slog.HandlerOptions{Level: slog.LevelWarn})),
|
|
||||||
}
|
|
||||||
|
|
||||||
announceManyPrefixes(h.table, routetable.MaxWaitingOriginChanges+1)
|
|
||||||
|
|
||||||
h.flushBatch()
|
|
||||||
|
|
||||||
var warning struct {
|
|
||||||
Msg string `json:"msg"`
|
|
||||||
Dropped *int `json:"dropped"`
|
|
||||||
}
|
|
||||||
|
|
||||||
found := false
|
|
||||||
|
|
||||||
for line := range bytes.SplitSeq(bytes.TrimSpace(logs.Bytes()), []byte("\n")) {
|
|
||||||
warning.Dropped = nil
|
|
||||||
|
|
||||||
err := json.Unmarshal(line, &warning)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("want a JSON log line, got %q: %v", line, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
if warning.Msg != "Origin changes dropped" {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
found = true
|
|
||||||
|
|
||||||
if warning.Dropped == nil || *warning.Dropped != 1 {
|
|
||||||
t.Errorf("the warning's dropped = %v, want 1", warning.Dropped)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if !found {
|
|
||||||
t.Errorf("no warning logged the dropped origin changes")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
+19
-75
@@ -88,10 +88,6 @@ const (
|
|||||||
ipVersionV6 = 6
|
ipVersionV6 = 6
|
||||||
)
|
)
|
||||||
|
|
||||||
// maxOriginChangesOnPage is the most origin changes the prefix page shows. A
|
|
||||||
// prefix with more shows the newest ones and says so.
|
|
||||||
const maxOriginChangesOnPage = 5000
|
|
||||||
|
|
||||||
// HealthCheckResponse represents the health check response.
|
// HealthCheckResponse represents the health check response.
|
||||||
type HealthCheckResponse struct {
|
type HealthCheckResponse struct {
|
||||||
Status string `json:"status"`
|
Status string `json:"status"`
|
||||||
@@ -1538,23 +1534,26 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
|
|||||||
|
|
||||||
routes := prefixRoutes(parsed, origins)
|
routes := prefixRoutes(parsed, origins)
|
||||||
|
|
||||||
changes, moreChanges, err := s.prefixOriginChanges(r.Context(), parsed)
|
|
||||||
if deadlinePassed(r.Context()) {
|
|
||||||
http.Error(w, requestTimeoutMessage, http.StatusGatewayTimeout)
|
|
||||||
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
if err != nil {
|
|
||||||
s.logger.Error("Failed to get the origin changes", "error", err)
|
|
||||||
http.Error(w, internalErrorMessage, http.StatusInternalServerError)
|
|
||||||
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// Prepare template data
|
// Prepare template data
|
||||||
data := newPrefixDetailData(parsed, routes, originInfos, changes,
|
data := struct {
|
||||||
moreChanges, cspNonce(r.Context()))
|
Prefix string
|
||||||
|
MaskLength int
|
||||||
|
IPVersion int
|
||||||
|
Routes []EnhancedRoute
|
||||||
|
Origins []*ASNInfo
|
||||||
|
PeerCount int
|
||||||
|
OriginCount int
|
||||||
|
CSPNonce string
|
||||||
|
}{
|
||||||
|
Prefix: parsed.String(),
|
||||||
|
MaskLength: parsed.Bits(),
|
||||||
|
IPVersion: ipVersionOf(parsed),
|
||||||
|
Routes: routesWithASPathHandles(routes),
|
||||||
|
Origins: originInfos,
|
||||||
|
PeerCount: len(routes),
|
||||||
|
OriginCount: len(originInfos),
|
||||||
|
CSPNonce: cspNonce(r.Context()),
|
||||||
|
}
|
||||||
|
|
||||||
// Check if context is still valid before writing response
|
// Check if context is still valid before writing response
|
||||||
select {
|
select {
|
||||||
@@ -1607,61 +1606,6 @@ func (s *Server) originASNs(
|
|||||||
return infos, nil
|
return infos, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// prefixOriginChanges returns prefix's origin changes for the page, newest
|
|
||||||
// first, at most maxOriginChangesOnPage, and whether there are more.
|
|
||||||
func (s *Server) prefixOriginChanges(
|
|
||||||
ctx context.Context, prefix netip.Prefix,
|
|
||||||
) ([]routetable.OriginChange, bool, error) {
|
|
||||||
// One more is read to know whether there are more.
|
|
||||||
changes, err := s.db.GetOriginChanges(ctx, prefix, maxOriginChangesOnPage+1)
|
|
||||||
if err != nil {
|
|
||||||
return nil, false, err
|
|
||||||
}
|
|
||||||
|
|
||||||
more := len(changes) > maxOriginChangesOnPage
|
|
||||||
if more {
|
|
||||||
changes = changes[:maxOriginChangesOnPage]
|
|
||||||
}
|
|
||||||
|
|
||||||
return changes, more, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// prefixDetailData is what the prefix detail page is rendered with.
|
|
||||||
type prefixDetailData struct {
|
|
||||||
Prefix string
|
|
||||||
MaskLength int
|
|
||||||
IPVersion int
|
|
||||||
Routes []EnhancedRoute
|
|
||||||
Origins []*ASNInfo
|
|
||||||
PeerCount int
|
|
||||||
OriginCount int
|
|
||||||
OriginChanges []routetable.OriginChange
|
|
||||||
MoreOriginChanges bool
|
|
||||||
// MaxOriginChanges is the most origin changes the page shows.
|
|
||||||
MaxOriginChanges int
|
|
||||||
CSPNonce string
|
|
||||||
}
|
|
||||||
|
|
||||||
// newPrefixDetailData builds the prefix detail page's data.
|
|
||||||
func newPrefixDetailData(
|
|
||||||
parsed netip.Prefix, routes []LiveRoute, originInfos []*ASNInfo,
|
|
||||||
changes []routetable.OriginChange, moreChanges bool, nonce string,
|
|
||||||
) prefixDetailData {
|
|
||||||
return prefixDetailData{
|
|
||||||
Prefix: parsed.String(),
|
|
||||||
MaskLength: parsed.Bits(),
|
|
||||||
IPVersion: ipVersionOf(parsed),
|
|
||||||
Routes: routesWithASPathHandles(routes),
|
|
||||||
Origins: originInfos,
|
|
||||||
PeerCount: len(routes),
|
|
||||||
OriginCount: len(originInfos),
|
|
||||||
OriginChanges: changes,
|
|
||||||
MoreOriginChanges: moreChanges,
|
|
||||||
MaxOriginChanges: maxOriginChangesOnPage,
|
|
||||||
CSPNonce: nonce,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// prefixRoutes returns one route for each peer that carries prefix, by origin
|
// prefixRoutes returns one route for each peer that carries prefix, by origin
|
||||||
// AS in the order of origins. Each origin's kept routes come first, newest
|
// AS in the order of origins. Each origin's kept routes come first, newest
|
||||||
// first, with their AS path, next hop and time. Its other peers follow without
|
// first, with their AS path, next hop and time. Its other peers follow without
|
||||||
|
|||||||
@@ -1,230 +0,0 @@
|
|||||||
package server
|
|
||||||
|
|
||||||
import (
|
|
||||||
"fmt"
|
|
||||||
"net/http"
|
|
||||||
"net/http/httptest"
|
|
||||||
"net/netip"
|
|
||||||
"strings"
|
|
||||||
"testing"
|
|
||||||
|
|
||||||
"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/metrics"
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/ristypes"
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/routetable"
|
|
||||||
"git.eeqj.de/sneak/routewatch/internal/streamer"
|
|
||||||
)
|
|
||||||
|
|
||||||
// testUncarriedASN is an AS the test's routing table does not carry for the prefix.
|
|
||||||
const testUncarriedASN = 64501
|
|
||||||
|
|
||||||
// newPrefixDetailServer returns a server whose routing table holds testPrefixV4
|
|
||||||
// with origin AS 64500, and whose database holds the given origin changes.
|
|
||||||
func newPrefixDetailServer(
|
|
||||||
t *testing.T, changes []routetable.OriginChange,
|
|
||||||
) *Server {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
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() })
|
|
||||||
|
|
||||||
table := emptyTable()
|
|
||||||
announce(table, testPrefixV4, 64500, testPeer(1), testRouteTime(0))
|
|
||||||
|
|
||||||
err = db.WriteOriginChanges(changes)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("WriteOriginChanges: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return New(db, table, streamer.New(logger.New(), metrics.New()),
|
|
||||||
logger.New(), &config.Config{})
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixDetailShowsOriginChanges checks that the prefix page shows the
|
|
||||||
// origin changes newest first, with the announced and withdrawn rows and the
|
|
||||||
// AS link. The rows use an AS the routing table does not carry for the
|
|
||||||
// prefix, so the link comes from the origin changes table.
|
|
||||||
func TestPrefixDetailShowsOriginChanges(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
prefix := netip.MustParsePrefix(testPrefixV4)
|
|
||||||
|
|
||||||
s := newPrefixDetailServer(t, []routetable.OriginChange{
|
|
||||||
{Prefix: prefix, OriginASN: testUncarriedASN, Announced: true,
|
|
||||||
PeerCount: 1, Time: testRouteTime(1)},
|
|
||||||
{Prefix: prefix, OriginASN: testUncarriedASN, Announced: false,
|
|
||||||
PeerCount: 0, Time: testRouteTime(2)},
|
|
||||||
})
|
|
||||||
|
|
||||||
rec := httptest.NewRecorder()
|
|
||||||
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
|
|
||||||
t.Context(), http.MethodGet, "/prefix/"+testPrefixV4, nil))
|
|
||||||
|
|
||||||
if rec.Code != http.StatusOK {
|
|
||||||
t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
|
|
||||||
}
|
|
||||||
|
|
||||||
page := rec.Body.String()
|
|
||||||
|
|
||||||
withdrawn := strings.Index(page, "Withdrawn")
|
|
||||||
announced := strings.Index(page, "Announced")
|
|
||||||
|
|
||||||
if withdrawn < 0 || announced < 0 {
|
|
||||||
t.Fatalf("page does not show the announced and withdrawn rows")
|
|
||||||
}
|
|
||||||
|
|
||||||
// The withdrawn row is newer, so it comes first.
|
|
||||||
if withdrawn > announced {
|
|
||||||
t.Errorf("the withdrawn row is not shown before the announced row")
|
|
||||||
}
|
|
||||||
|
|
||||||
if !strings.Contains(page, `href="/as/64501"`) {
|
|
||||||
t.Errorf("page does not link to AS64501")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixDetailShowsOriginZeroAsNoKnownOrigin checks that the origin
|
|
||||||
// changes table shows AS 0 as no known origin, without a link to AS0.
|
|
||||||
func TestPrefixDetailShowsOriginZeroAsNoKnownOrigin(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
prefix := netip.MustParsePrefix(testPrefixV4)
|
|
||||||
|
|
||||||
s := newPrefixDetailServer(t, []routetable.OriginChange{
|
|
||||||
{Prefix: prefix, OriginASN: ristypes.NoOrigin, Announced: true,
|
|
||||||
PeerCount: 1, Time: testRouteTime(1)},
|
|
||||||
})
|
|
||||||
|
|
||||||
rec := httptest.NewRecorder()
|
|
||||||
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
|
|
||||||
t.Context(), http.MethodGet, "/prefix/"+testPrefixV4, nil))
|
|
||||||
|
|
||||||
if rec.Code != http.StatusOK {
|
|
||||||
t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
|
|
||||||
}
|
|
||||||
|
|
||||||
page := rec.Body.String()
|
|
||||||
|
|
||||||
if !strings.Contains(page, "No known origin") {
|
|
||||||
t.Errorf("page does not say the origin change has no known origin")
|
|
||||||
}
|
|
||||||
|
|
||||||
if strings.Contains(page, `href="/as/0"`) {
|
|
||||||
t.Errorf("page links to AS0")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixDetailSaysWhenThereAreMoreChangesThanItShows checks that the
|
|
||||||
// prefix page says so when a prefix has more origin changes than the page's
|
|
||||||
// limit.
|
|
||||||
func TestPrefixDetailSaysWhenThereAreMoreChangesThanItShows(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
prefix := netip.MustParsePrefix(testPrefixV4)
|
|
||||||
|
|
||||||
changes := make([]routetable.OriginChange, 0, maxOriginChangesOnPage+1)
|
|
||||||
for i := range maxOriginChangesOnPage + 1 {
|
|
||||||
changes = append(changes, routetable.OriginChange{
|
|
||||||
Prefix: prefix, OriginASN: 64500, Announced: true, PeerCount: 1,
|
|
||||||
Time: testRouteTime(i),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
s := newPrefixDetailServer(t, changes)
|
|
||||||
|
|
||||||
rec := httptest.NewRecorder()
|
|
||||||
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
|
|
||||||
t.Context(), http.MethodGet, "/prefix/"+testPrefixV4, nil))
|
|
||||||
|
|
||||||
if rec.Code != http.StatusOK {
|
|
||||||
t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
|
|
||||||
}
|
|
||||||
|
|
||||||
page := rec.Body.String()
|
|
||||||
|
|
||||||
note := fmt.Sprintf("more than %d changes", maxOriginChangesOnPage)
|
|
||||||
if !strings.Contains(page, note) {
|
|
||||||
t.Errorf("page does not say there are more than %d changes",
|
|
||||||
maxOriginChangesOnPage)
|
|
||||||
}
|
|
||||||
|
|
||||||
if got := strings.Count(page, "<td>Announced</td>"); got != maxOriginChangesOnPage {
|
|
||||||
t.Errorf("page shows %d change rows, want %d", got,
|
|
||||||
maxOriginChangesOnPage)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixDetailShowsNoNoteAtTheLimit checks that a prefix with exactly the
|
|
||||||
// page's limit of origin changes shows all of them and no note.
|
|
||||||
func TestPrefixDetailShowsNoNoteAtTheLimit(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
prefix := netip.MustParsePrefix(testPrefixV4)
|
|
||||||
|
|
||||||
changes := make([]routetable.OriginChange, 0, maxOriginChangesOnPage)
|
|
||||||
for i := range maxOriginChangesOnPage {
|
|
||||||
changes = append(changes, routetable.OriginChange{
|
|
||||||
Prefix: prefix, OriginASN: 64500, Announced: true, PeerCount: 1,
|
|
||||||
Time: testRouteTime(i),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
s := newPrefixDetailServer(t, changes)
|
|
||||||
|
|
||||||
rec := httptest.NewRecorder()
|
|
||||||
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
|
|
||||||
t.Context(), http.MethodGet, "/prefix/"+testPrefixV4, nil))
|
|
||||||
|
|
||||||
if rec.Code != http.StatusOK {
|
|
||||||
t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
|
|
||||||
}
|
|
||||||
|
|
||||||
page := rec.Body.String()
|
|
||||||
|
|
||||||
if strings.Contains(page, "more than") {
|
|
||||||
t.Errorf("page says there are more changes, but a prefix has %d",
|
|
||||||
maxOriginChangesOnPage)
|
|
||||||
}
|
|
||||||
|
|
||||||
if got := strings.Count(page, "<td>Announced</td>"); got != maxOriginChangesOnPage {
|
|
||||||
t.Errorf("page shows %d change rows, want %d", got,
|
|
||||||
maxOriginChangesOnPage)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestPrefixDetailShowsOriginChangeTimeAndPeerCount checks that a row shows
|
|
||||||
// its time and its peer count.
|
|
||||||
func TestPrefixDetailShowsOriginChangeTimeAndPeerCount(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
prefix := netip.MustParsePrefix(testPrefixV4)
|
|
||||||
|
|
||||||
s := newPrefixDetailServer(t, []routetable.OriginChange{
|
|
||||||
{Prefix: prefix, OriginASN: 64500, Announced: true, PeerCount: 7,
|
|
||||||
Time: testRouteTime(5)},
|
|
||||||
})
|
|
||||||
|
|
||||||
rec := httptest.NewRecorder()
|
|
||||||
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
|
|
||||||
t.Context(), http.MethodGet, "/prefix/"+testPrefixV4, nil))
|
|
||||||
|
|
||||||
if rec.Code != http.StatusOK {
|
|
||||||
t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
|
|
||||||
}
|
|
||||||
|
|
||||||
page := rec.Body.String()
|
|
||||||
|
|
||||||
if !strings.Contains(page, "2026-01-02 03:09:05") {
|
|
||||||
t.Errorf("page does not show the origin change's time")
|
|
||||||
}
|
|
||||||
|
|
||||||
if !strings.Contains(page, "<td>7</td>") {
|
|
||||||
t.Errorf("page does not show the origin change's peer count")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -7,8 +7,8 @@
|
|||||||
// - the AS name search
|
// - the AS name search
|
||||||
//
|
//
|
||||||
// Every route it shows comes from the routing table in memory. AS names,
|
// Every route it shows comes from the routing table in memory. AS names,
|
||||||
// WHOIS details, the times a prefix and an AS were first seen, and a prefix's
|
// WHOIS details and the times a prefix and an AS were first seen come from
|
||||||
// origin changes come from the database.
|
// the database.
|
||||||
package server
|
package server
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
|||||||
@@ -299,41 +299,6 @@
|
|||||||
</div>
|
</div>
|
||||||
{{end}}
|
{{end}}
|
||||||
|
|
||||||
{{if .OriginChanges}}
|
|
||||||
<div class="origins-section">
|
|
||||||
<h3>Origin Changes</h3>
|
|
||||||
{{if .MoreOriginChanges}}
|
|
||||||
<p>There are more than {{.MaxOriginChanges}} changes. Showing the newest {{.MaxOriginChanges}}.</p>
|
|
||||||
{{end}}
|
|
||||||
<table class="route-table">
|
|
||||||
<thead>
|
|
||||||
<tr>
|
|
||||||
<th>Time (UTC)</th>
|
|
||||||
<th>Change</th>
|
|
||||||
<th>Origin AS</th>
|
|
||||||
<th>Peers</th>
|
|
||||||
</tr>
|
|
||||||
</thead>
|
|
||||||
<tbody>
|
|
||||||
{{range .OriginChanges}}
|
|
||||||
<tr>
|
|
||||||
<td>{{.Time.Format "2006-01-02 15:04:05"}}</td>
|
|
||||||
<td>{{if .Announced}}Announced{{else}}Withdrawn{{end}}</td>
|
|
||||||
<td>
|
|
||||||
{{if .OriginASN}}
|
|
||||||
<a href="/as/{{.OriginASN}}" class="as-link">AS{{.OriginASN}}</a>
|
|
||||||
{{else}}
|
|
||||||
No known origin
|
|
||||||
{{end}}
|
|
||||||
</td>
|
|
||||||
<td>{{.PeerCount}}</td>
|
|
||||||
</tr>
|
|
||||||
{{end}}
|
|
||||||
</tbody>
|
|
||||||
</table>
|
|
||||||
</div>
|
|
||||||
{{end}}
|
|
||||||
|
|
||||||
{{if .Routes}}
|
{{if .Routes}}
|
||||||
<div class="routes-section">
|
<div class="routes-section">
|
||||||
<div class="routes-header">
|
<div class="routes-header">
|
||||||
|
|||||||
Reference in New Issue
Block a user