diff --git a/README.md b/README.md index 19c3bbc..bb61bf2 100644 --- a/README.md +++ b/README.md @@ -142,7 +142,11 @@ 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 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 - prefix's traffic to. RouteWatch records the peer itself as the next hop. + prefix's traffic to. RouteWatch records the peer itself as the next hop. The + page also lists the prefix's origin changes, newest first: each time an origin + AS gained its first peer or lost its last, with the time, announced or + withdrawn, the origin AS and the peer count. It shows at most the newest + 5,000. - `/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 route. When peers give a prefix different origin ASes, the page shows the one @@ -516,7 +520,8 @@ 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, 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 - uses. With much less, it would run almost all the time. + uses. With much less, it would run almost all the time. The routing table also + 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 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 @@ -638,7 +643,7 @@ The parts: - 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 reclaims free space every 10 minutes. Every hour it refreshes the statistics - SQLite uses to plan its queries. + SQLite uses to plan its queries, and prunes the origin changes table. - The HTTP server serves the pages and the API. Every route it shows comes from the routing table in memory. WHOIS details and first-seen times come from the database. AS names come from the database or from the AS list built into @@ -809,6 +814,10 @@ routes(prefix, origin_asn, peers, -- The peers of the routes table, each with its number in the peer sets 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, UNIQUE(as_a, as_b)) bgp_peers(id PRIMARY KEY, peer_ip UNIQUE, peer_asn, first_seen, last_seen, diff --git a/TODO.md b/TODO.md index 54beb9d..e3353ad 100644 --- a/TODO.md +++ b/TODO.md @@ -41,9 +41,8 @@ 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 `9bd0e91` on the live feed is checking whether any handler still drops updates. -Two questions wait for sneak: which history the prefix page keeps -(https://git.eeqj.de/sneak/routewatch/issues/161), and which fields the IP -lookup answer gains (https://git.eeqj.de/sneak/routewatch/issues/203). +One question waits for sneak: 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 https://git.eeqj.de/sneak/routewatch/issues/86. @@ -57,6 +56,13 @@ problem the run finds. # Completed Steps +- 2026-11-05: the prefix page lists the prefix's origin changes, newest first: + each time an origin AS gained its first peer or lost its last, with the time, + announced or withdrawn, the origin AS and the peer count. 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 and 72 hours. The page shows at most the newest 5,000 (closes + #161) - 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 the queue was half full. The prefix handler's queue now holds 100,000 diff --git a/internal/database/interface.go b/internal/database/interface.go index 0274ebb..717d26a 100644 --- a/internal/database/interface.go +++ b/internal/database/interface.go @@ -55,6 +55,14 @@ type Store interface { WriteRouteChanges(changes []routetable.Change, newPeers map[string]uint) 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 GetNextStaleASN(ctx context.Context, staleThreshold time.Duration) (int, error) UpdateASNWHOIS(ctx context.Context, update *ASNWHOISUpdate) error diff --git a/internal/database/originchanges.go b/internal/database/originchanges.go new file mode 100644 index 0000000..6967d64 --- /dev/null +++ b/internal/database/originchanges.go @@ -0,0 +1,172 @@ +package database + +import ( + "context" + "fmt" + "net/netip" + "time" + + "git.eeqj.de/sneak/routewatch/internal/routetable" +) + +// maxOriginChangesPerPrefix is how many origin changes the prune keeps for +// each prefix: the newest ones. A prefix with more has its older rows deleted. +const maxOriginChangesPerPrefix = 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 +// maxOriginChangesPerPrefix. +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 +// maxOriginChangesPerPrefix rows, the rows older than originChangesRetention +// that are not among its newest maxOriginChangesPerPrefix. 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, + maxOriginChangesPerPrefix) + 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, maxOriginChangesPerPrefix-1) + + d.unlock() + + if err != nil { + return fmt.Errorf("failed to prune origin changes of %s: %w", + prefix, err) + } + } + + return nil +} diff --git a/internal/database/originchanges_internal_test.go b/internal/database/originchanges_internal_test.go new file mode 100644 index 0000000..9404f16 --- /dev/null +++ b/internal/database/originchanges_internal_test.go @@ -0,0 +1,160 @@ +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 every row newer than 72 +// hours, keeps the newest 500 rows of each prefix, deletes the rest, and +// leaves other prefixes alone. +func TestOriginChangesPrune(t *testing.T) { + t.Parallel() + + db := newRoutesTestDatabase(t) + oldPrefix := netip.MustParsePrefix(tablePrefixV4) + newPrefix := netip.MustParsePrefix(tablePrefixShortV4) + other := netip.MustParsePrefix("10.20.0.0/16") + + now := time.Now() + + changes := make([]routetable.OriginChange, 0, 1201) + + // 600 rows for oldPrefix, all older than 72 hours, oldest first. The + // prune keeps the newest 500 and deletes the oldest 100. + for i := range 600 { + changes = append(changes, routetable.OriginChange{ + Prefix: oldPrefix, OriginASN: asA, Announced: true, PeerCount: 1, + Time: now.Add(-100 * time.Hour).Add(time.Duration(i) * time.Minute), + }) + } + + // 600 rows for newPrefix, all newer than 72 hours, oldest first. The + // prune keeps all of them. + for i := range 600 { + changes = append(changes, routetable.OriginChange{ + Prefix: newPrefix, OriginASN: asA, Announced: true, PeerCount: 1, + Time: now.Add(-71 * time.Hour).Add(time.Duration(i) * time.Minute), + }) + } + + // One row for other, which the prune leaves alone. + changes = append(changes, routetable.OriginChange{ + Prefix: other, OriginASN: asA, Announced: true, PeerCount: 1, + Time: now.Add(-100 * time.Hour), + }) + + 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) + } + + gotOld, err := db.GetOriginChanges(t.Context(), oldPrefix, 1000) + if err != nil { + t.Fatalf("GetOriginChanges: %v", err) + } + + if len(gotOld) != 500 { + t.Errorf("GetOriginChanges(oldPrefix) = %d rows, want 500", len(gotOld)) + } + + gotNew, err := db.GetOriginChanges(t.Context(), newPrefix, 1000) + if err != nil { + t.Fatalf("GetOriginChanges: %v", err) + } + + if len(gotNew) != 600 { + t.Errorf("GetOriginChanges(newPrefix) = %d rows, want 600", len(gotNew)) + } + + 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)) + } +} + +// 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)) + } +} diff --git a/internal/database/schema.sql b/internal/database/schema.sql index 8abd176..edb556a 100644 --- a/internal/database/schema.sql +++ b/internal/database/schema.sql @@ -119,4 +119,19 @@ CREATE TABLE IF NOT EXISTS routes ( CREATE TABLE IF NOT EXISTS route_peers ( number INTEGER PRIMARY KEY, address TEXT NOT NULL UNIQUE -); \ No newline at end of file +); + +-- 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 and 72 +-- hours. 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); \ No newline at end of file diff --git a/internal/routetable/originchanges.go b/internal/routetable/originchanges.go new file mode 100644 index 0000000..c1361cc --- /dev/null +++ b/internal/routetable/originchanges.go @@ -0,0 +1,88 @@ +package routetable + +import ( + "net/netip" + "slices" + "time" +) + +// maxWaitingOriginChanges is the most origin changes the table keeps waiting +// for the prefix handler to write. A 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(), + } + + if len(t.originChanges) == maxWaitingOriginChanges { + // The oldest change is dropped to make room. + t.originChanges = t.originChanges[1:] + t.originChangesDropped++ + } + + t.originChanges = append(t.originChanges, change) +} + +// 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. +func (t *Table) PutBackOriginChanges(changes []OriginChange) { + t.mu.Lock() + defer t.mu.Unlock() + + t.originChanges = append(changes, 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 +} diff --git a/internal/routetable/originchanges_test.go b/internal/routetable/originchanges_test.go new file mode 100644 index 0000000..b6b97ed --- /dev/null +++ b/internal/routetable/originchanges_test.go @@ -0,0 +1,224 @@ +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() + + const limit = 100000 + + table := routetable.New(routetable.Params{}) + addr := netip.MustParseAddr("10.0.0.0") + + for range limit + 1 { + prefix := netip.PrefixFrom(addr, 32) + table.Announce(prefix, asA, routeOf(peerA, asA, 10)) + + addr = addr.Next() + } + + changes, dropped := table.TakeOriginChanges(limit) + if dropped != 1 { + t.Errorf("dropped = %d, want 1", dropped) + } + + if len(changes) != limit { + t.Errorf("len(changes) = %d, want %d", len(changes), limit) + } +} + +// 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) + } +} diff --git a/internal/routetable/routetable.go b/internal/routetable/routetable.go index 339df90..983d7a3 100644 --- a/internal/routetable/routetable.go +++ b/internal/routetable/routetable.go @@ -2,7 +2,9 @@ // 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 // announcements, withdrawals and peers' sessions going down. It answers the -// route reads that the pages and the API need. +// route reads that the pages and the API need. It also records each prefix and +// origin AS pair's origin changes: when it gains its first peer and when it +// loses its last. package routetable import ( @@ -91,6 +93,9 @@ type Params struct { // peer not in it gets the next number above all of them the first time it // announces a route. 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. @@ -126,11 +131,24 @@ type Table struct { // originsWithoutKeptRoute counts the origins that peers carry but that // keep no route. 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 // params.PeerNumbers. func New(params Params) *Table { + now := params.Now + if now == nil { + now = time.Now + } + t := &Table{ ipv4: newFamily(ipv4Bits, minLookupMaskLengthV4), ipv6: newFamily(ipv6Bits, minLookupMaskLengthV6), @@ -139,6 +157,7 @@ func New(params Params) *Table { peerAddresses: make(map[uint]string, len(params.PeerNumbers)), newPeers: make(map[string]uint), changed: make(map[pair]struct{}), + now: now, keptTimes: btree.NewG(keptTimesDegree, func(a, b keptTime) bool { return a.time.Before(b.time) }), @@ -469,6 +488,8 @@ func (t *Table) join(prefix netip.Prefix, o *origin, r route) { wasWithoutKeptRoute := o.withoutKeptRoute() joined := !o.peers.Test(r.peer) + + firstPeer := joined && o.peers.None() if joined { o.peers.Set(r.peer) @@ -482,6 +503,10 @@ func (t *Table) join(prefix netip.Prefix, o *origin, r route) { if joined || dropped || kept { 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 @@ -549,6 +574,7 @@ func (t *Table) recordChange( t.countOriginWithoutKeptRoute(o, wasWithoutKeptRoute) if o.peers.None() { + t.recordOriginChange(prefix, o.asn, false, t.prefixPeerCount(prefix)) t.removeOrigin(prefix, o) } } diff --git a/internal/routewatch/app_integration_internal_test.go b/internal/routewatch/app_integration_internal_test.go index c145e9b..3a0e6a3 100644 --- a/internal/routewatch/app_integration_internal_test.go +++ b/internal/routewatch/app_integration_internal_test.go @@ -330,6 +330,23 @@ func (m *mockStore) ReadRouteTable(_ context.Context) (*routetable.Table, error) 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 func (m *mockStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error { m.mu.Lock() diff --git a/internal/routewatch/dbmaintainer.go b/internal/routewatch/dbmaintainer.go index abd075c..1384253 100644 --- a/internal/routewatch/dbmaintainer.go +++ b/internal/routewatch/dbmaintainer.go @@ -25,6 +25,10 @@ const ( // analyzeInterval is how often to run ANALYZE. 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 = 10 * time.Second @@ -33,6 +37,9 @@ const ( // analyzeTimeout is the max time for ANALYZE. 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 @@ -111,6 +118,9 @@ func (m *DBMaintainer) run(ctx context.Context) { analyzeTimer := time.NewTimer(analyzeInterval) defer analyzeTimer.Stop() + pruneTimer := time.NewTimer(originChangesPruneInterval) + defer pruneTimer.Stop() + for { select { case <-m.stopCh: @@ -127,6 +137,10 @@ func (m *DBMaintainer) run(ctx context.Context) { case <-analyzeTimer.C: m.runAnalyze(ctx) analyzeTimer.Reset(analyzeInterval) + + case <-pruneTimer.C: + m.runPruneOriginChanges(ctx) + pruneTimer.Reset(originChangesPruneInterval) } } } @@ -175,6 +189,18 @@ 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 // returns how long it took and its error. func runWithTimeout( diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 3de5114..bb36a32 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -48,6 +48,10 @@ const ( // going down changes are written in about a minute. 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 // 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 @@ -507,6 +511,7 @@ func (h *PrefixHandler) takeBatch() []seenPrefix { // writes. func (h *PrefixHandler) flushBatch() { h.writeTableChanges() + h.writeOriginChanges() batch := h.takeBatch() if len(batch) == 0 { @@ -608,6 +613,30 @@ func (h *PrefixHandler) writeTableChanges() { } } +// writeOriginChanges writes at most maxOriginChangesPerWrite 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() { + changes, dropped := h.table.TakeOriginChanges(maxOriginChangesPerWrite) + + if dropped > 0 { + h.logger.Warn("Origin changes dropped", "dropped", dropped) + } + + if len(changes) == 0 { + return + } + + err := h.db.WriteOriginChanges(changes) + 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 // the prefix tables store and the IP lookup reads them in. The feed sends IPv6 // announcements compressed ("2001:db8::/32") but IPv6 withdrawals uncompressed diff --git a/internal/routewatch/prefixhandler_originchanges_internal_test.go b/internal/routewatch/prefixhandler_originchanges_internal_test.go new file mode 100644 index 0000000..7bc88ab --- /dev/null +++ b/internal/routewatch/prefixhandler_originchanges_internal_test.go @@ -0,0 +1,84 @@ +package routewatch + +import ( + "errors" + "net/netip" + "testing" + + "git.eeqj.de/sneak/routewatch/internal/database" + "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) + } +} diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 8ae2d90..be9d897 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -88,6 +88,10 @@ const ( 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. type HealthCheckResponse struct { Status string `json:"status"` @@ -1534,27 +1538,24 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc { routes := prefixRoutes(parsed, origins) - // Prepare template data - data := struct { - 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()), + 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 + data := newPrefixDetailData(parsed, routes, originInfos, changes, + moreChanges, cspNonce(r.Context())) + // Check if context is still valid before writing response select { case <-r.Context().Done(): @@ -1606,6 +1607,58 @@ func (s *Server) originASNs( 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 + 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, + CSPNonce: nonce, + } +} + // 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 // first, with their AS path, next hop and time. Its other peers follow without diff --git a/internal/server/prefix_originchanges_internal_test.go b/internal/server/prefix_originchanges_internal_test.go new file mode 100644 index 0000000..58dbbb7 --- /dev/null +++ b/internal/server/prefix_originchanges_internal_test.go @@ -0,0 +1,148 @@ +package server + +import ( + "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" +) + +// 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. +func TestPrefixDetailShowsOriginChanges(t *testing.T) { + t.Parallel() + + prefix := netip.MustParsePrefix(testPrefixV4) + + s := newPrefixDetailServer(t, []routetable.OriginChange{ + {Prefix: prefix, OriginASN: 64500, Announced: true, PeerCount: 1, + Time: testRouteTime(1)}, + {Prefix: prefix, OriginASN: 64500, 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/64500"`) { + t.Errorf("page does not link to AS64500") + } +} + +// 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") + } +} + +// TestPrefixDetailSaysWhenMoreThan5000Changes checks that the prefix page says +// so when a prefix has more than 5,000 origin changes. +func TestPrefixDetailSaysWhenMoreThan5000Changes(t *testing.T) { + t.Parallel() + + prefix := netip.MustParsePrefix(testPrefixV4) + + changes := make([]routetable.OriginChange, 0, 5001) + for i := range 5001 { + 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 5,000 changes") { + t.Errorf("page does not say there are more than 5,000 changes") + } +} diff --git a/internal/templates/prefix_detail.html b/internal/templates/prefix_detail.html index de44b3f..a1f8824 100644 --- a/internal/templates/prefix_detail.html +++ b/internal/templates/prefix_detail.html @@ -297,6 +297,41 @@ {{end}} +{{end}} + + {{if .OriginChanges}} +
+

Origin Changes

+ {{if .MoreOriginChanges}} +

There are more than 5,000 changes. Showing the newest 5,000.

+ {{end}} + + + + + + + + + + + {{range .OriginChanges}} + + + + + + + {{end}} + +
Time (UTC)ChangeOrigin ASPeers
{{.Time.Format "2006-01-02 15:04:05"}}{{if .Announced}}Announced{{else}}Withdrawn{{end}} + {{if .OriginASN}} + AS{{.OriginASN}} + {{else}} + No known origin + {{end}} + {{.PeerCount}}
+
{{end}} {{if .Routes}}