diff --git a/Dockerfile b/Dockerfile index f2f9d83..5351025 100644 --- a/Dockerfile +++ b/Dockerfile @@ -139,8 +139,13 @@ EXPOSE 8080 COPY ./entrypoint.sh /entrypoint.sh -# Health check using the health endpoint, on the port PORT names -HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ +# Health check using the health endpoint, on the port PORT names. During the +# start period, Docker 25 and later checks every 5 seconds, and the first pass +# makes the container healthy. After it, Docker checks every 30 seconds. The +# start period covers the 60 seconds after which upaas reads the health +# (README "Running under upaas"). Startup reads the routing table first, +# which takes about 25 seconds when the table is full. +HEALTHCHECK --interval=30s --timeout=5s --start-period=60s --retries=3 \ CMD curl -sf "http://localhost:${PORT:-8080}/.well-known/healthcheck.json" || exit 1 ENTRYPOINT ["/bin/bash", "/entrypoint.sh" ] diff --git a/README.md b/README.md index e0cf0ef..c907cfa 100644 --- a/README.md +++ b/README.md @@ -285,6 +285,8 @@ routewatch/ 2. **Initialization** (`internal/routewatch/app.go`) - Database created with SQLite schema (WAL mode, page caches as in "Memory") + - The routing table read from the database into memory, before the streamer + connects - Message handlers registered with the streamer - HTTP server started on configured port @@ -311,7 +313,10 @@ routewatch/ holds AS 0 anywhere, deletes all of a peer's routes when its BGP session goes down). Routes are kept per peer IP, so when one peer IP has sessions on two collectors, a session going down on one also removes the routes the - other still carries, until they are announced again. + other still carries, until they are announced again. It also applies each + update to the routing table in memory, and writes the table's changes to + the database every second. A write takes at most 20,000 changed prefix and + origin AS pairs, and the rest follow in the next writes. - **PeeringHandler**: Extracts AS peering relationships from AS paths 5. **HTTP Request Flow** @@ -495,7 +500,9 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app needs: - Health check: the image's `HEALTHCHECK` requests `/.well-known/healthcheck.json` on the container port. upaas reads the container's health 60 seconds after a deploy and fails the deploy unless it is - `healthy`. + `healthy`. For the first 60 seconds, Docker 25 and later runs the check every + 5 seconds, and the first pass makes the container healthy. Startup first reads + the routing table, which takes about 25 seconds when the table is full. - Stop grace: the container needs 9 seconds between SIGTERM and SIGKILL. upaas gives 10 (`stopTimeoutSeconds` in its `internal/docker/client.go`), as `docker stop` does by default. The daemon allows 8 seconds for the handlers to diff --git a/TODO.md b/TODO.md index e293558..f805dcb 100644 --- a/TODO.md +++ b/TODO.md @@ -33,6 +33,12 @@ tested. # Completed Steps +- 2026-10-27: the prefix handler applies each update to the routing table in + memory, and writes the table's changes to the database every second. A write + takes at most 20,000 changed pairs, and the rest follow in the next writes. A + failed write is written again later. Startup reads the table before the + streamer connects. The pages still read the per-peer route tables (part of + #122) - 2026-10-27: the database keeps a copy of the whole routing table in a new `routes` table, one row per prefix and origin AS. The peers' numbers are in a new `route_peers` table. One function writes the pairs that changed in memory, @@ -254,5 +260,5 @@ tested. # Future Steps - the rest of https://git.eeqj.de/sneak/routewatch/issues/122, one PR each: the - prefix handler writing to the routing table, the pages and the API reading - from it, then removing the per-peer route tables + pages and the API reading from the routing table, then removing the per-peer + route tables diff --git a/internal/database/interface.go b/internal/database/interface.go index d36f8ea..1e8bf5f 100644 --- a/internal/database/interface.go +++ b/internal/database/interface.go @@ -7,6 +7,8 @@ package database import ( "context" "time" + + "git.eeqj.de/sneak/routewatch/internal/routetable" ) // Stats contains database statistics including counts for ASNs, prefixes, @@ -69,6 +71,10 @@ type Store interface { GetLiveRouteCounts() (ipv4Count, ipv6Count int, err error) GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error) + // Routing table operations: the database's copy of the table in memory + WriteRouteChanges(changes []routetable.Change, newPeers map[string]uint) error + ReadRouteTable(ctx context.Context) (*routetable.Table, error) + // IP lookup operations GetIPInfo(ip string) (*IPInfo, error) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error) diff --git a/internal/database/routes_internal_test.go b/internal/database/routes_internal_test.go index 14a70d9..79bc133 100644 --- a/internal/database/routes_internal_test.go +++ b/internal/database/routes_internal_test.go @@ -3,6 +3,7 @@ package database import ( "errors" "fmt" + "math" "math/rand/v2" "net/netip" "os" @@ -71,7 +72,7 @@ func newRoutesTestDatabase(t *testing.T) *Database { func writeTableChanges(t *testing.T, db *Database, table *routetable.Table) { t.Helper() - changes, newPeers := table.TakeChanges() + changes, newPeers := table.TakeChanges(math.MaxInt) err := db.WriteRouteChanges(changes, newPeers) if err != nil { @@ -178,7 +179,7 @@ func checkReadTable( checkSameReads(t, got, want) - changes, newPeers := got.TakeChanges() + changes, newPeers := got.TakeChanges(math.MaxInt) if len(changes) != 0 || len(newPeers) != 0 { t.Errorf("the read recorded changes %+v and new peers %v, want none", changes, newPeers) @@ -369,7 +370,7 @@ func TestWriteRouteChangesWritesNothingWhenItFails(t *testing.T) { announceTo(table, tablePrefixV6, asA, peerB, 20) // peerB's number is missing, so its kept route cannot be written. - changes, newPeers := table.TakeChanges() + changes, newPeers := table.TakeChanges(math.MaxInt) delete(newPeers, peerB) err := db.WriteRouteChanges(changes, newPeers) diff --git a/internal/routetable/benchmark_test.go b/internal/routetable/benchmark_test.go index 002d391..be4f2c4 100644 --- a/internal/routetable/benchmark_test.go +++ b/internal/routetable/benchmark_test.go @@ -2,6 +2,7 @@ package routetable_test import ( "fmt" + "math" "math/rand/v2" "net/netip" "runtime" @@ -77,7 +78,7 @@ func BenchmarkApplyMixedBatch(b *testing.B) { // The record of changes is handed over every second while routewatch // runs, so it is not part of the table's size. - table.TakeChanges() + table.TakeChanges(math.MaxInt) bytesPerPair := float64(heapBytes()-before) / benchPrefixes diff --git a/internal/routetable/reads.go b/internal/routetable/reads.go index fb925d7..e108984 100644 --- a/internal/routetable/reads.go +++ b/internal/routetable/reads.go @@ -31,6 +31,11 @@ type OriginatedPrefix struct { // Counts are the table's figures for the stats. type Counts struct { + // Pairs counts the prefix and origin AS pairs that some peer carries. + Pairs int + // Peers counts the peers that have a number: those given in + // Params.PeerNumbers and those numbered since. + Peers int IPv4Prefixes int IPv6Prefixes int // IPv4PrefixesByLength[n] and IPv6PrefixesByLength[n] are how many @@ -160,6 +165,7 @@ func (t *Table) Counts() Counts { defer t.mu.RUnlock() counts := Counts{ + Peers: len(t.peerNumbers), IPv4PrefixesByLength: t.ipv4.prefixesByLength(), IPv6PrefixesByLength: t.ipv6.prefixesByLength(), IPv4Routes: t.ipv4.routes, @@ -175,6 +181,10 @@ func (t *Table) Counts() Counts { counts.IPv6Prefixes += n } + for _, prefixes := range t.byOrigin { + counts.Pairs += len(prefixes) + } + oldest, found := t.keptTimes.Min() if found { counts.OldestKeptRoute = &oldest.time diff --git a/internal/routetable/reads_test.go b/internal/routetable/reads_test.go index f4e20a4..c3f8288 100644 --- a/internal/routetable/reads_test.go +++ b/internal/routetable/reads_test.go @@ -194,6 +194,8 @@ func TestCounts(t *testing.T) { announceAll(table, "2001:db8::/32", asA, 5, peerA) announceAll(table, "2001:db8::/32", asA, 25, peerC) checkCounts(t, table, routetable.Counts{ + Pairs: 3, + Peers: 3, IPv4Prefixes: 2, IPv6Prefixes: 1, IPv4PrefixesByLength: byLength(ipv4Bits, map[int]int{16: 1, 24: 1}), @@ -209,6 +211,8 @@ func TestCounts(t *testing.T) { announceAll(table, prefixV4, asA, 30, peerC) table.Withdraw(netip.MustParsePrefix("2001:db8::/32"), peerA) checkCounts(t, table, routetable.Counts{ + Pairs: 3, + Peers: 3, IPv4Prefixes: 2, IPv6Prefixes: 1, IPv4PrefixesByLength: byLength(ipv4Bits, map[int]int{16: 1, 24: 1}), @@ -219,9 +223,11 @@ func TestCounts(t *testing.T) { NewestKeptRoute: timeAt(30), }) - // peerC was the last peer of the IPv6 prefix. + // peerC was the last peer of the IPv6 prefix. It keeps its number. table.PeerDown(peerC) checkCounts(t, table, routetable.Counts{ + Pairs: 2, + Peers: 3, IPv4Prefixes: 2, IPv4PrefixesByLength: byLength(ipv4Bits, map[int]int{16: 1, 24: 1}), IPv6PrefixesByLength: byLength(ipv6Bits, nil), diff --git a/internal/routetable/routetable.go b/internal/routetable/routetable.go index 416f7c2..2a737c5 100644 --- a/internal/routetable/routetable.go +++ b/internal/routetable/routetable.go @@ -7,6 +7,7 @@ package routetable import ( "cmp" + "maps" "net/netip" "slices" "sync" @@ -102,11 +103,15 @@ type Table struct { peerNumbers map[string]uint peerAddresses map[uint]string nextPeerNumber uint - // peerPrefixes counts the prefixes each peer carries, by its number. - peerPrefixes map[uint]int + // peerPrefixes[n] counts the prefixes the peer numbered n carries, for + // each n below nextPeerNumber. It is a slice, not a map, because Load + // counts once for every peer of every pair it loads: with a map, reading + // a full table at startup took several times as long. + peerPrefixes []int - // newPeers and changed are what TakeChanges hands over next: the peers - // numbered, and the prefix and origin AS pairs changed, since it last ran. + // newPeers and changed are what TakeChanges hands over: the peers + // numbered since it last ran, and the prefix and origin AS pairs changed + // since it last took them. newPeers map[string]uint changed map[pair]struct{} @@ -127,7 +132,6 @@ func New(params Params) *Table { byOrigin: make(map[int]map[netip.Prefix]*origin), peerNumbers: make(map[string]uint, len(params.PeerNumbers)), peerAddresses: make(map[uint]string, len(params.PeerNumbers)), - peerPrefixes: make(map[uint]int), newPeers: make(map[string]uint), changed: make(map[pair]struct{}), keptTimes: btree.NewG(keptTimesDegree, func(a, b keptTime) bool { @@ -141,6 +145,8 @@ func New(params Params) *Table { t.nextPeerNumber = max(t.nextPeerNumber, number+1) } + t.peerPrefixes = make([]int, t.nextPeerNumber) + return t } @@ -240,13 +246,14 @@ func (t *Table) Load(change Change) { o := t.findOrAddOrigin(prefix, change.OriginASN) wasWithoutKeptRoute := o.withoutKeptRoute() - for peer := range change.Peers.EachSet() { - if !o.peers.Test(peer) { - o.peers.Set(peer) + // The peers not yet in the peer set join it in one union, which takes + // less time than setting their bits one at a time. + joining := change.Peers.Difference(&o.peers) + o.peers.InPlaceUnion(joining) + t.familyOf(prefix.Addr()).routes += int(joining.Count()) - t.familyOf(prefix.Addr()).routes++ - t.peerPrefixes[peer]++ - } + for peer := range joining.EachSet() { + t.peerPrefixes[peer]++ } // Oldest first, as they were announced, so that two routes with the same @@ -265,16 +272,29 @@ func (t *Table) Load(change Change) { t.countOriginWithoutKeptRoute(o, wasWithoutKeptRoute) } -// TakeChanges returns the prefix and origin AS pairs that changed since it last -// ran, in prefix order, each with its current content. It also returns the -// peers numbered since then, by address. It then clears both records. -func (t *Table) TakeChanges() ([]Change, map[string]uint) { +// TakeChanges returns at most maxChanges of the prefix and origin AS pairs that +// changed since they were last taken, in prefix order, each with its current +// content. The other changed pairs wait for a later call. It also returns every +// peer numbered since it last ran, by address, so a pair taken later names only +// peers already returned. It then forgets what it returned. +func (t *Table) TakeChanges(maxChanges int) ([]Change, map[string]uint) { t.mu.Lock() defer t.mu.Unlock() - changes := make([]Change, 0, len(t.changed)) + changes := make([]Change, 0, min(maxChanges, len(t.changed))) for key := range t.changed { + if len(changes) == maxChanges { + break + } + changes = append(changes, t.currentContent(key)) + delete(t.changed, key) + } + + // A map keeps the memory of its largest size, so an emptied one is + // replaced. + if len(t.changed) == 0 { + t.changed = make(map[pair]struct{}) } slices.SortFunc(changes, func(a, b Change) int { @@ -284,11 +304,35 @@ func (t *Table) TakeChanges() ([]Change, map[string]uint) { newPeers := t.newPeers t.newPeers = make(map[string]uint) - t.changed = make(map[pair]struct{}) return changes, newPeers } +// PutBackChanges records again what one TakeChanges call returned, for a +// caller that could not write it. The next TakeChanges returns those peers +// again, and those pairs wait to be taken again like any changed pair. Each is +// then taken with its content at that time, so a later change of a pair +// replaces the earlier. +func (t *Table) PutBackChanges(changes []Change, newPeers map[string]uint) { + t.mu.Lock() + defer t.mu.Unlock() + + for _, change := range changes { + t.changed[pair{prefix: change.Prefix, asn: change.OriginASN}] = struct{}{} + } + + maps.Copy(t.newPeers, newPeers) +} + +// ChangeCount returns how many changed prefix and origin AS pairs wait for +// TakeChanges. +func (t *Table) ChangeCount() int { + t.mu.RLock() + defer t.mu.RUnlock() + + return len(t.changed) +} + // numberPeer returns the peer's number, and gives it the next one the first // time the table sees it. func (t *Table) numberPeer(address string) uint { @@ -301,6 +345,7 @@ func (t *Table) numberPeer(address string) uint { t.nextPeerNumber++ t.peerNumbers[address] = number t.peerAddresses[number] = address + t.peerPrefixes = append(t.peerPrefixes, 0) t.newPeers[address] = number return number diff --git a/internal/routetable/routetable_test.go b/internal/routetable/routetable_test.go index 9251674..d77f15f 100644 --- a/internal/routetable/routetable_test.go +++ b/internal/routetable/routetable_test.go @@ -1,6 +1,7 @@ package routetable_test import ( + "math" "net/netip" "reflect" "slices" @@ -438,7 +439,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) { // peerB to v4 are kept; peerA keeps only its bit. announceAll(table, prefixV4, asA, 10, peerA, peerB, peerC) table.Announce(v6, asA, routeOf(peerB, asA, 10)) - table.TakeChanges() + table.TakeChanges(math.MaxInt) // peerA announces again, with a time older than both kept routes. table.Announce(v4, asA, routeOf(peerA, asA, 15)) @@ -476,6 +477,80 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) { checkChanges(t, table, []change{}, map[string]uint{}) } +// TestTakeChangesTakesAtMostMaxChanges changes three prefix and origin AS +// pairs, then takes the changes two at a time. The first call must return two +// pairs and both new peers, and the second the third pair and no peer. +func TestTakeChangesTakesAtMostMaxChanges(t *testing.T) { + t.Parallel() + + const maxChanges = 2 + + table := routetable.New(routetable.Params{}) + v4 := netip.MustParsePrefix(prefixV4) + v6 := netip.MustParsePrefix(prefixV6) + + table.Announce(v4, asA, routeOf(peerA, asA, 10)) + table.Announce(v4, asB, routeOf(peerB, asB, 10)) + table.Announce(v6, asA, routeOf(peerA, asA, 10)) + + first, firstPeers := table.TakeChanges(maxChanges) + second, secondPeers := table.TakeChanges(maxChanges) + + if len(first) != maxChanges || len(second) != 1 { + t.Fatalf("TakeChanges(%d) returned %d and then %d pairs, want %d and 1", + maxChanges, len(first), len(second), maxChanges) + } + + wantPeers := map[string]uint{peerA: 0, peerB: 1} + if !reflect.DeepEqual(firstPeers, wantPeers) || len(secondPeers) != 0 { + t.Errorf("TakeChanges(%d) returned the new peers %v and then %v, "+ + "want %v and none", maxChanges, firstPeers, secondPeers, wantPeers) + } + + for _, c := range first { + if c.Prefix == second[0].Prefix && c.OriginASN == second[0].OriginASN { + t.Errorf("both calls returned %s with AS%d", c.Prefix, c.OriginASN) + } + } +} + +// TestPutBackChanges takes a table's changes and puts them back, as a caller +// does whose write failed. The next TakeChanges must return them again, with +// each pair's content then, and with the changes since. +func TestPutBackChanges(t *testing.T) { + t.Parallel() + + table := routetable.New(routetable.Params{}) + 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.PutBackChanges(table.TakeChanges(math.MaxInt)) + + if got := table.ChangeCount(); got != 2 { + t.Errorf("ChangeCount() = %d after the changes were put back, want 2", + got) + } + + // peerA withdraws v4, so v4's earlier content must not come back; peerB + // announces v6, and is a new peer as well. + table.Withdraw(v4, peerA) + table.Announce(v6, asA, routeOf(peerB, asA, 20)) + + checkChanges(t, table, []change{ + {Prefix: prefixV4, OriginASN: asA, Removed: true}, + {Prefix: prefixV6, OriginASN: asA, Peers: []uint{0, 1}, + Routes: []routetable.Route{ + routeOf(peerB, asA, 20), routeOf(peerA, asA, 10), + }}, + }, map[string]uint{peerA: 0, peerB: 1}) + + if got := table.ChangeCount(); got != 0 { + t.Errorf("ChangeCount() = %d after TakeChanges, want 0", got) + } +} + func TestLoad(t *testing.T) { t.Parallel() @@ -497,7 +572,7 @@ func TestLoad(t *testing.T) { source.Withdraw(other, peerB) source.Withdraw(other, peerC) - changes, peers := source.TakeChanges() + changes, peers := source.TakeChanges(math.MaxInt) loaded := routetable.New(routetable.Params{PeerNumbers: peers}) // Loading a pair twice changes nothing. @@ -543,7 +618,8 @@ func TestConcurrentUse(t *testing.T) { func(round int) { table.Announce(prefix, asB, routeOf(peerB, asB, round)) }, func(int) { table.Withdraw(prefix, peerA) }, func(int) { table.PeerDown(peerB) }, - func(int) { table.TakeChanges() }, + func(int) { table.PutBackChanges(table.TakeChanges(math.MaxInt)) }, + func(int) { table.ChangeCount() }, func(int) { table.Lookup(prefix.Addr()) }, func(int) { table.Origins(prefix) }, func(int) { table.OriginatedPrefixes(asA) }, @@ -563,14 +639,15 @@ func TestConcurrentUse(t *testing.T) { wg.Wait() } -// checkChanges checks that TakeChanges returns want and wantPeers. +// checkChanges checks that TakeChanges, taking every change, returns want and +// wantPeers. func checkChanges( t *testing.T, table *routetable.Table, want []change, wantPeers map[string]uint, ) { t.Helper() - changes, newPeers := table.TakeChanges() + changes, newPeers := table.TakeChanges(math.MaxInt) got := make([]change, 0, len(changes)) for _, c := range changes { diff --git a/internal/routewatch/app.go b/internal/routewatch/app.go index 9fa7d20..a8ee4bd 100644 --- a/internal/routewatch/app.go +++ b/internal/routewatch/app.go @@ -8,6 +8,7 @@ package routewatch import ( "context" "errors" + "fmt" "path/filepath" "sync" "time" @@ -17,6 +18,7 @@ import ( "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/routetable" "git.eeqj.de/sneak/routewatch/internal/server" "git.eeqj.de/sneak/routewatch/internal/streamer" @@ -92,6 +94,14 @@ func (rw *RouteWatch) Run(ctx context.Context) error { // Register database handler to process BGP UPDATE messages if rw.config.EnableBatchedDatabaseWrites { rw.logger.Info("Using batched database handlers for improved performance") + + // The routing table is read before the streamer connects, so that + // the prefix handler applies the feed's updates to all of it. + table, err := rw.readRouteTable(ctx) + if err != nil { + return err + } + // ASHandler maintains the asns table rw.dbHandler = NewASHandler(rw.db, rw.logger) rw.streamer.RegisterHandler(rw.dbHandler) @@ -100,8 +110,9 @@ func (rw *RouteWatch) Run(ctx context.Context) error { rw.peerHandler = NewPeerHandler(rw.db, rw.logger) rw.streamer.RegisterHandler(rw.peerHandler) - // PrefixHandler maintains the prefixes and live_routes tables - rw.prefixHandler = NewPrefixHandler(rw.db, rw.logger) + // PrefixHandler maintains the prefixes and live_routes tables, and + // the routing table in memory and its copy in the routes table + rw.prefixHandler = NewPrefixHandler(rw.db, rw.logger, table) rw.prefixHandler.SetMetricsTracker(rw.streamer.GetMetricsTracker()) rw.streamer.RegisterHandler(rw.prefixHandler) @@ -254,6 +265,27 @@ func (rw *RouteWatch) closeDatabase(ctx context.Context) { } } +// readRouteTable reads the routing table from the database and logs how many +// prefix and origin AS pairs and peers it holds and how long the read took. +// ctx is fx's start context, which ends 15 seconds after the start. Reading a +// full table can take longer, so the read does not stop with ctx. +func (rw *RouteWatch) readRouteTable( + ctx context.Context, +) (*routetable.Table, error) { + start := time.Now() + + table, err := rw.db.ReadRouteTable(context.WithoutCancel(ctx)) + if err != nil { + return nil, fmt.Errorf("failed to read the routing table: %w", err) + } + + counts := table.Counts() + rw.logger.Info("Read the routing table", "pairs", counts.Pairs, + "peers", counts.Peers, "duration_ms", time.Since(start).Milliseconds()) + + return table, nil +} + // getModule provides all fx dependencies func getModule() []fx.Option { return []fx.Option{ diff --git a/internal/routewatch/app_integration_internal_test.go b/internal/routewatch/app_integration_internal_test.go index ad4f18a..86c78bb 100644 --- a/internal/routewatch/app_integration_internal_test.go +++ b/internal/routewatch/app_integration_internal_test.go @@ -19,6 +19,7 @@ import ( "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/server" "git.eeqj.de/sneak/routewatch/internal/streamer" ) @@ -39,6 +40,9 @@ type mockStore struct { Prefixes map[string]*database.Prefix Peerings map[string]bool // key is "from_to" Routes map[string]bool // key is "prefix_origin_peer" + // TablePrefixes holds the prefixes that the written changes of the + // routing table leave some peer carrying. + TablePrefixes map[string]bool // Track IP versions IPv4Prefixes int @@ -48,10 +52,11 @@ type mockStore struct { // newMockStore creates a new mock store func newMockStore() *mockStore { return &mockStore{ - ASNs: make(map[int]*database.ASN), - Prefixes: make(map[string]*database.Prefix), - Peerings: make(map[string]bool), - Routes: make(map[string]bool), + ASNs: make(map[int]*database.ASN), + Prefixes: make(map[string]*database.Prefix), + Peerings: make(map[string]bool), + Routes: make(map[string]bool), + TablePrefixes: make(map[string]bool), } } @@ -395,6 +400,30 @@ func (m *mockStore) DeleteLiveRoutesOfPeer(_ string) (int, error) { return 0, nil } +// WriteRouteChanges records the prefixes the changes leave some peer carrying, +// whatever their origin AS. +func (m *mockStore) WriteRouteChanges( + changes []routetable.Change, _ map[string]uint, +) error { + m.mu.Lock() + defer m.mu.Unlock() + + for _, change := range changes { + if change.Removed { + delete(m.TablePrefixes, change.Prefix.String()) + } else { + m.TablePrefixes[change.Prefix.String()] = true + } + } + + return nil +} + +// ReadRouteTable returns an empty routing table. +func (m *mockStore) ReadRouteTable(_ context.Context) (*routetable.Table, error) { + return routetable.New(routetable.Params{}), nil +} + // GetOrCreateASNBatch mock implementation func (m *mockStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error { m.mu.Lock() @@ -566,7 +595,8 @@ func announcementFeed(prefixes []string) string { // handler's worker waits on the first one until the server sees the streamer // close the connection, which only Shutdown does, so all three are still in // the streamer when Shutdown starts. Each prefix must be in the database when -// Shutdown returns. +// Shutdown returns, both in the prefix tables and in the routing table's +// writes. func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) { t.Parallel() @@ -600,7 +630,7 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) { Logger: log, Config: cfg, }) - rw.prefixHandler = NewPrefixHandler(store, log) + rw.prefixHandler = NewPrefixHandler(store, log, routetable.New(routetable.Params{})) s.RegisterHandler(gatedPrefixHandler{ PrefixHandler: rw.prefixHandler, release: release, }) @@ -640,9 +670,11 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) { defer store.mu.Unlock() for _, prefix := range prefixes { - if _, ok := store.Prefixes[prefix]; !ok { - t.Errorf("%s was queued in the streamer when Shutdown started "+ - "but is not in the database", prefix) + _, inPrefixTables := store.Prefixes[prefix] + if !inPrefixTables || !store.TablePrefixes[prefix] { + t.Errorf("%s was queued in the streamer when Shutdown started; "+ + "written to the prefix tables %t, in the routing table's "+ + "writes %t", prefix, inPrefixTables, store.TablePrefixes[prefix]) } } } diff --git a/internal/routewatch/batchwrite_internal_test.go b/internal/routewatch/batchwrite_internal_test.go index b9ce48d..42c16bc 100644 --- a/internal/routewatch/batchwrite_internal_test.go +++ b/internal/routewatch/batchwrite_internal_test.go @@ -13,6 +13,7 @@ import ( "git.eeqj.de/sneak/routewatch/internal/database" "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/ristypes" + "git.eeqj.de/sneak/routewatch/internal/routetable" ) const ( @@ -119,10 +120,11 @@ func (s *blockingStore) releaseWrites() { // //nolint:tagliatelle // field names are the keys the handlers log type stopLog struct { - Msg string `json:"msg"` - Updates int `json:"updates"` - PeersDown int `json:"peers_down"` - Paths int `json:"paths"` + Msg string `json:"msg"` + Updates int `json:"updates"` + PeersDown int `json:"peers_down"` + TableChanges int `json:"table_changes"` + Paths int `json:"paths"` } // newErrorLogger returns a logger that writes only errors, as JSON lines, to @@ -313,6 +315,9 @@ func stopWhileAWriteIsBlocked( } } +// TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline also checks the +// routing table's changes: the flush whose delete is held wrote those of the +// fill, so Stop must log the one of the extra prefix as not written. func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { t.Parallel() @@ -321,7 +326,8 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { var logs bytes.Buffer store := newBlockingStore() - h := NewPrefixHandler(store, newErrorLogger(&logs)) + h := NewPrefixHandler(store, newErrorLogger(&logs), + routetable.New(routetable.Params{})) fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) // The peer's session goes down first, so the write held is the delete of // its routes, and Stop must log that peer down too. @@ -337,9 +343,51 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { } logged := lastLogged(t, &logs) - if logged.Updates != prefixBatchSize+1 || logged.PeersDown != 1 { - t.Errorf("Stop logged %d updates and %d peer downs not written, "+ - "want %d and 1", logged.Updates, logged.PeersDown, prefixBatchSize+1) + if logged.Updates != prefixBatchSize+1 || logged.PeersDown != 1 || + logged.TableChanges != 1 { + t.Errorf("Stop logged %d updates, %d peer downs and %d table changes "+ + "not written, want %d, 1 and 1", logged.Updates, logged.PeersDown, + logged.TableChanges, prefixBatchSize+1) + } +} + +// tableWriteBlockingStore is a blockingStore whose writes of the routing +// table's changes also wait until the test calls releaseWrites. +type tableWriteBlockingStore struct { + *blockingStore +} + +func (s tableWriteBlockingStore) WriteRouteChanges( + changes []routetable.Change, newPeers map[string]uint, +) error { + s.waitForRelease() + + return s.mockStore.WriteRouteChanges(changes, newPeers) +} + +// TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress holds the write +// 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 +// meanwhile. Stop must log as not written both the change the held write took +// and the one since. +func TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress(t *testing.T) { + t.Parallel() + + const extraPrefix = "198.51.100.0/24" + + var logs bytes.Buffer + + blocking := newBlockingStore() + h := NewPrefixHandler(tableWriteBlockingStore{blockingStore: blocking}, + newErrorLogger(&logs), routetable.New(routetable.Params{})) + fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) + stopWhileAWriteIsBlocked(t, blocking, h, h.done, + []*ristypes.RISMessage{prefixAnnouncement(fill)}, + prefixAnnouncement([]string{extraPrefix})) + + if logged := lastLogged(t, &logs); logged.TableChanges != 2 { + t.Errorf("Stop logged %d table changes not written, want 2", + logged.TableChanges) } } @@ -455,7 +503,7 @@ func TestPrefixHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) { const extraPrefix = "198.51.100.0/24" store := newBlockingStore() - h := NewPrefixHandler(store, logger.New()) + h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{})) fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) handleWhileAWriteIsBlocked(t, store, h, []*ristypes.RISMessage{prefixAnnouncement(fill)}, @@ -515,7 +563,7 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) { const extraPrefix = "198.51.100.0/24" store := newBlockingStore() - h := NewPrefixHandler(store, logger.New()) + h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{})) fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchMaxSize) handleAtMaxBatchSize(t, store, h, @@ -592,7 +640,10 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) { name: "prefix", batchSize: prefixBatchSize, handle: func(n int) chan struct{} { - h := &PrefixHandler{batchFullCh: make(chan struct{}, 1)} + h := &PrefixHandler{ + batchFullCh: make(chan struct{}, 1), + table: routetable.New(routetable.Params{}), + } h.HandleMessage(prefixAnnouncement( slices.Repeat([]string{testV4Prefix}, n))) diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index 8a468fd..c2764cf 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -13,6 +13,7 @@ import ( "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" ) const ( @@ -38,6 +39,16 @@ const ( // DO NOT reduce this timeout - larger batches are more efficient prefixBatchTimeout = 1 * time.Second + // maxTableChangesPerWrite is the most changed prefix and origin AS pairs + // of the routing table that one flush writes; the rest wait for the next + // flushes. The change of one pair takes about 430 bytes of memory and 15 + // microseconds to write. So a write holds about 9 MB and takes about a + // third of the one-second flush interval, which leaves the rest to the + // batch. The feed peaks at about 5,000 updates a second, each changing at + // most one pair. So the million pairs that a full-table peer's session + // going down changes are written in about a minute. + maxTableChangesPerWrite = 20000 + // IP version constants ipv4Version = 4 ipv6Version = 6 @@ -47,15 +58,19 @@ const ( // database. Routes are added on announcement and deleted on withdrawal or on an // announcement whose AS path is empty or holds AS 0 anywhere, and all routes of // a peer are deleted when its BGP session goes down. +// +// It applies the same updates to the routing table in memory as it handles +// them, and writes the table's changes to the database with each flush. type PrefixHandler struct { db database.Store logger *logger.Logger metrics *metrics.Tracker + table *routetable.Table - // Batching. mu guards batch, held, peersDown, writing and deleting only: - // the flush goroutine takes them under mu and writes them after releasing - // mu, so HandleMessage waits for a database write only while batch is at - // prefixBatchMaxSize. + // Batching. mu guards batch, held, peersDown, writing, deleting and + // writingTableChanges only: the flush goroutine takes them under mu and + // writes them after releasing mu, so HandleMessage waits for a database + // write only while batch is at prefixBatchMaxSize. mu sync.Mutex batch []prefixUpdate // held holds the updates that the peers whose route delete failed in the @@ -74,6 +89,9 @@ type PrefixHandler struct { // deleting is how many peers from peersDown the flush goroutine took and // has not yet deleted the routes of. deleting int + // writingTableChanges is how many changes the flush goroutine took from + // the routing table and has not finished writing. + writingTableChanges int // batchTaken is signalled when the flush goroutine takes batch; its L is // &mu. batchTaken sync.Cond @@ -97,11 +115,26 @@ type prefixUpdate struct { endsInSet bool } -// NewPrefixHandler creates a new batched prefix tracking handler -func NewPrefixHandler(db database.Store, logger *logger.Logger) *PrefixHandler { +// announcesRoute says whether u gives the peer a route to the prefix: an +// announcement whose path gives an origin, or ends in an AS set of several +// members, which gives origin 0 (none). Any other update removes the peer's +// route for the prefix: a withdrawal, or an announcement whose path is empty +// or holds AS 0 anywhere (the decoder empties such a path). Such an +// announcement replaces the peer's route, as any announcement does, with none +// (RFC 7607 treats AS 0 in a path as a withdrawal). +func (u prefixUpdate) announcesRoute() bool { + return u.messageType == "announcement" && (u.originASN > 0 || u.endsInSet) +} + +// NewPrefixHandler creates a new batched prefix tracking handler, which applies +// the updates to table, the routing table read from db at startup. +func NewPrefixHandler( + db database.Store, logger *logger.Logger, table *routetable.Table, +) *PrefixHandler { h := &PrefixHandler{ db: db, logger: logger, + table: table, batch: make([]prefixUpdate, 0, prefixBatchSize), batchFullCh: make(chan struct{}, 1), lastFlush: time.Now(), @@ -165,18 +198,22 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { h.batchTaken.Wait() } - // Withdrawals are queued before announcements, because the batch keeps - // each peer's last update for a prefix: a message that lists a prefix as - // both withdrawn and announced keeps the route (RFC 4271 section 4.3). - // A withdrawal names only the prefix and peer: the message's AS path, and - // so its origin AS, belongs to the announced prefixes. + // Withdrawals are queued, and applied to the routing table, before + // announcements, because the batch keeps each peer's last update for a + // prefix: a message that lists a prefix as both withdrawn and announced + // keeps the route (RFC 4271 section 4.3). A withdrawal names only the + // prefix and peer: the message's AS path, and so its origin AS, belongs + // to the announced prefixes. for _, prefix := range msg.Withdrawals { - h.batch = append(h.batch, prefixUpdate{ + update := prefixUpdate{ prefix: canonicalPrefix(prefix), peer: msg.Peer, messageType: "withdrawal", timestamp: timestamp, - }) + } + h.batch = append(h.batch, update) + h.applyToTable(update) + // Record withdrawal in metrics if h.metrics != nil { h.metrics.RecordWithdrawal() @@ -185,7 +222,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { for _, announcement := range msg.Announcements { for _, prefix := range announcement.Prefixes { - h.batch = append(h.batch, prefixUpdate{ + update := prefixUpdate{ prefix: canonicalPrefix(prefix), originASN: originASN, peer: msg.Peer, @@ -193,7 +230,10 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { timestamp: timestamp, path: path, endsInSet: len(msg.Path.Set) > 0, - }) + } + h.batch = append(h.batch, update) + h.applyToTable(update) + // Record announcement in metrics if h.metrics != nil { h.metrics.RecordAnnouncement() @@ -212,12 +252,15 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { } } -// Stop stops the flush goroutine, which writes the updates and peer downs the -// handler holds before it returns, and waits for it until ctx ends. Once ctx -// has ended the flush goroutine skips that last write, and a write still -// running is lost when the process exits. A peer whose route delete fails in -// the last write is not retried either. Stop logs how many updates are not -// written and how many peer downs have not had their routes deleted. +// Stop stops the flush goroutine, which makes one last flush before it returns, +// and waits for it until ctx ends. That flush writes the updates and peer downs +// the handler holds, and at most maxTableChangesPerWrite of the routing table's +// changes. Once ctx has ended the flush goroutine skips that last flush, and a +// write still running is lost when the process exits. A peer whose route delete +// fails in the last flush is not retried either, nor is a failed write of the +// table's changes. Stop logs how many updates are not written, how many peer +// downs have not had their routes deleted, and how many of the table's changes +// are not written. func (h *PrefixHandler) Stop(ctx context.Context) { h.stopCh <- ctx @@ -229,22 +272,24 @@ func (h *PrefixHandler) Stop(ctx context.Context) { h.mu.Lock() unwritten := h.writing + len(h.batch) + len(h.held) peersDown := h.deleting + len(h.peersDown) + tableChanges := h.writingTableChanges + h.table.ChangeCount() h.mu.Unlock() - if unwritten == 0 && peersDown == 0 { + if unwritten == 0 && peersDown == 0 && tableChanges == 0 { return } if ctx.Err() != nil { h.logger.Error("Stop deadline passed before the prefix handler wrote "+ - "its updates", "updates", unwritten, "peers_down", peersDown) + "its updates", "updates", unwritten, "peers_down", peersDown, + "table_changes", tableChanges) return } - h.logger.Error("The prefix handler stopped without deleting the routes "+ - "of peers whose session went down", "updates", unwritten, - "peers_down", peersDown) + h.logger.Error("The prefix handler stopped before it wrote everything", + "updates", unwritten, "peers_down", peersDown, + "table_changes", tableChanges) } // peerSessionDown notes that the peer's BGP session went down, so that the next @@ -252,11 +297,14 @@ func (h *PrefixHandler) Stop(ctx context.Context) { // peer's updates still in the batch or held came before the down, which // replaces them, so they are dropped; updates the peer sends later stay in the // batch and are written after the delete. A peer already in peersDown is not -// added again: one delete removes all of its routes. +// added again: one delete removes all of its routes. In the routing table in +// memory the down takes effect at once. func (h *PrefixHandler) peerSessionDown(peer string) { h.mu.Lock() defer h.mu.Unlock() + h.table.PeerDown(peer) + isPeers := func(update prefixUpdate) bool { return update.peer == peer } h.batch = slices.DeleteFunc(h.batch, isPeers) h.held = slices.DeleteFunc(h.held, isPeers) @@ -321,9 +369,11 @@ func (h *PrefixHandler) takeBatch() ([]prefixUpdate, []string) { return batch, peersDown } -// flushBatch takes the pending batch and writes it to the database, without -// holding the mutex during the write. +// flushBatch writes the routing table's changes, then takes the pending batch +// and writes it to the database, without holding the mutex during the writes. func (h *PrefixHandler) flushBatch() { + h.writeTableChanges() + batch, peersDown := h.takeBatch() if len(batch) == 0 && len(peersDown) == 0 { return @@ -370,15 +420,7 @@ func (h *PrefixHandler) flushBatch() { ) for _, update := range lastUpdates { - // An announcement is stored when its path gives an origin, or when it - // ends in an AS set of several members: that route is stored with - // origin 0 (none). Any other update deletes the peer's route for the - // prefix: a withdrawal, or an announcement whose path is empty or - // holds AS 0 anywhere (the decoder empties such a path). Such an - // announcement replaces the peer's route, as any announcement does, - // with none (RFC 7607 treats AS 0 in a path as a withdrawal). - if update.messageType == "announcement" && - (update.originASN > 0 || update.endsInSet) { + if update.announcesRoute() { // Create live route for batch upsert route := h.createLiveRoute(update) if route != nil { @@ -421,6 +463,59 @@ func (h *PrefixHandler) flushBatch() { h.lastFlush = time.Now() } +// applyToTable applies update to the routing table in memory: as an +// announcement when it gives the peer a route, otherwise as a withdrawal. A +// prefix that does not parse is left out; flushBatch logs it. +func (h *PrefixHandler) applyToTable(update prefixUpdate) { + prefix, err := netip.ParsePrefix(update.prefix) + if err != nil { + return + } + + if !update.announcesRoute() { + h.table.Withdraw(prefix, update.peer) + + return + } + + h.table.Announce(prefix, update.originASN, routetable.Route{ + Peer: update.peer, + ASPath: update.path, + Time: update.timestamp, + }) +} + +// writeTableChanges writes at most maxTableChangesPerWrite changes of the +// routing table in memory, and every peer the table numbered since the last +// write. +// When the write fails, they go back to the table, so that a later write writes +// them again, each pair with its content then. +func (h *PrefixHandler) writeTableChanges() { + h.mu.Lock() + changes, newPeers := h.table.TakeChanges(maxTableChangesPerWrite) + h.writingTableChanges = len(changes) + h.mu.Unlock() + + if len(changes) == 0 && len(newPeers) == 0 { + return + } + + err := h.db.WriteRouteChanges(changes, newPeers) + + h.mu.Lock() + defer h.mu.Unlock() + + h.writingTableChanges = 0 + + if err != nil { + h.logger.Error("Failed to write the routing table's changes; the next "+ + "flush retries", "changes", len(changes), "new_peers", len(newPeers), + "error", err) + + h.table.PutBackChanges(changes, newPeers) + } +} + // deleteLiveRoutesOfPeers deletes all live routes of each of peers, the peers // whose BGP session went down, and returns those whose delete failed. Most // downs come from peers that have no routes, so only a delete that removed diff --git a/internal/routewatch/prefixhandler_internal_test.go b/internal/routewatch/prefixhandler_internal_test.go index f047a65..45fb3b1 100644 --- a/internal/routewatch/prefixhandler_internal_test.go +++ b/internal/routewatch/prefixhandler_internal_test.go @@ -16,6 +16,7 @@ import ( "git.eeqj.de/sneak/routewatch/internal/database" "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/ristypes" + "git.eeqj.de/sneak/routewatch/internal/routetable" ) const testPeerIP = "2001:db8:ffff::1" @@ -40,9 +41,7 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) { t.Parallel() db := newTestDatabase(t) - - // Built without NewPrefixHandler's flush timer, so the test flushes itself. - h := &PrefixHandler{db: db, logger: logger.New()} + h := prefixHandlerWithoutTimer(db) ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) h.HandleMessage(&ristypes.RISMessage{ @@ -462,8 +461,7 @@ func TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse(t *testing.T) { t.Fatalf("failed to create database: %v", err) } - // Built without NewPrefixHandler's flush timer, so the test flushes itself. - h := &PrefixHandler{db: db, logger: logger.New()} + h := prefixHandlerWithoutTimer(db) withdrawal := &ristypes.RISMessage{ Type: ristypes.MessageTypeUpdate, @@ -694,8 +692,7 @@ func TestPrefixHandlerHandlesADownWhileAFlushWrites(t *testing.T) { release: make(chan struct{}), } - // Built without NewPrefixHandler's flush timer, so the test flushes itself. - h := &PrefixHandler{db: store, logger: logger.New()} + h := prefixHandlerWithoutTimer(store) h.HandleMessage(peerAnnouncement(testPeerIP, beforeDown)) flushed := make(chan struct{}) @@ -832,10 +829,7 @@ func TestPrefixHandlerRetriesAFailedDeleteOfAPeersRoutes(t *testing.T) { db := newTestDatabase(t) store := &firstPeerDeleteFailsStore{Database: db} - - // Built without NewPrefixHandler's flush timer, so the test - // flushes itself. - h := &PrefixHandler{db: store, logger: logger.New()} + h := prefixHandlerWithoutTimer(store) store.whileFailing = func() { for _, msg := range tt.whileFailing { h.HandleMessage(msg) @@ -897,11 +891,7 @@ func TestPrefixHandlerKeepsAtMostABatchOfUpdatesWhileADeleteKeepsFailing( otherPeersPrefixes := []string{testV4Prefix, testV6Prefix, "2001:db8:2::/48"} db := newTestDatabase(t) - - // Built without NewPrefixHandler's flush timer, so the test flushes itself. - h := &PrefixHandler{ - db: peerDeleteAlwaysFailsStore{Database: db}, logger: logger.New(), - } + h := prefixHandlerWithoutTimer(peerDeleteAlwaysFailsStore{Database: db}) h.HandleMessage(sessionDown()) // The peer announces a /32 for each address from 10.0.0.0 on, each @@ -953,7 +943,7 @@ func TestPrefixHandlerStopLogsAFailedDeleteWithoutTheDeadline(t *testing.T) { var logs bytes.Buffer h := NewPrefixHandler(peerDeleteAlwaysFailsStore{Database: newTestDatabase(t)}, - newErrorLogger(&logs)) + newErrorLogger(&logs), routetable.New(routetable.Params{})) h.HandleMessage(sessionDown()) h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix)) h.Stop(t.Context()) @@ -979,7 +969,8 @@ func TestPrefixHandlerStopLogsNothingAfterADeleteThatSucceeds(t *testing.T) { var logs bytes.Buffer - h := NewPrefixHandler(newTestDatabase(t), newErrorLogger(&logs)) + h := NewPrefixHandler(newTestDatabase(t), newErrorLogger(&logs), + routetable.New(routetable.Params{})) h.HandleMessage(sessionDown()) h.Stop(t.Context()) @@ -989,6 +980,17 @@ func TestPrefixHandlerStopLogsNothingAfterADeleteThatSucceeds(t *testing.T) { } } +// prefixHandlerWithoutTimer returns a prefix handler that writes to store and +// applies the updates to an empty routing table. It is built without +// NewPrefixHandler's flush goroutine, so the test flushes itself. +func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler { + return &PrefixHandler{ + db: store, + logger: logger.New(), + table: routetable.New(routetable.Params{}), + } +} + // newTestDatabase opens a database in a new temporary directory. It is closed // when the test ends. func newTestDatabase(t *testing.T) *database.Database { @@ -1012,9 +1014,7 @@ func handleBatches( t.Helper() db := newTestDatabase(t) - - // Built without NewPrefixHandler's flush timer, so the test flushes itself. - h := &PrefixHandler{db: db, logger: logger.New()} + h := prefixHandlerWithoutTimer(db) for _, batch := range batches { for _, msg := range batch { diff --git a/internal/routewatch/prefixhandler_table_internal_test.go b/internal/routewatch/prefixhandler_table_internal_test.go new file mode 100644 index 0000000..4b64db3 --- /dev/null +++ b/internal/routewatch/prefixhandler_table_internal_test.go @@ -0,0 +1,535 @@ +package routewatch + +import ( + "bytes" + "encoding/json" + "errors" + "log/slog" + "net/netip" + "reflect" + "slices" + "testing" + "time" + + "git.eeqj.de/sneak/routewatch/internal/database" + "git.eeqj.de/sneak/routewatch/internal/logger" + "git.eeqj.de/sneak/routewatch/internal/ristypes" + "git.eeqj.de/sneak/routewatch/internal/routetable" +) + +// More peers and prefixes for the tests of the routing table in memory. +const ( + secondPeerIP = "2001:db8:ffff::2" + thirdPeerIP = "2001:db8:ffff::3" + fourthPeerIP = "192.0.2.4" + fifthPeerIP = "192.0.2.5" + + otherV4Prefix = "198.51.100.0/24" + setV4Prefix = "203.0.113.0/24" + + // firstManyPrefix is the first prefix announceManyPrefixes announces. + firstManyPrefix = "10.0.0.0/32" +) + +// tableWriteTimeout is how long a test waits for the flush timer to write the +// routing table's changes: a few ticks of it. +const tableWriteTimeout = 5 * prefixBatchTimeout + +// tableCase is one test of the routing table in memory: the messages handed to +// the prefix handler, and for each prefix the peers that carry it with each +// origin AS after them. +type tableCase struct { + name string + messages []*ristypes.RISMessage + want map[string]map[int][]string +} + +// TestPrefixHandlerAppliesUpdatesToTheRoutingTable hands the prefix handler +// announcements, withdrawals and a peer's session going down. +func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) { + t.Parallel() + + announceAndWithdraw := peerAnnouncement(testPeerIP, testV4Prefix) + announceAndWithdraw.Withdrawals = []string{testV4Prefix} + + checkTableCases(t, []tableCase{ + { + name: "an announcement", + messages: []*ristypes.RISMessage{peerAnnouncement(testPeerIP, testV4Prefix)}, + want: map[string]map[int][]string{testV4Prefix: {testASNB: {testPeerIP}}}, + }, + { + name: "a withdrawal", + messages: []*ristypes.RISMessage{ + peerAnnouncement(testPeerIP, testV4Prefix), + peerWithdrawal(testPeerIP, testV4Prefix), + }, + want: map[string]map[int][]string{testV4Prefix: {}}, + }, + { + // RFC 4271 section 4.3: the announcement counts, not the withdrawal. + name: "a message that announces and withdraws the prefix", + messages: []*ristypes.RISMessage{announceAndWithdraw}, + want: map[string]map[int][]string{testV4Prefix: {testASNB: {testPeerIP}}}, + }, + { + name: "a peer's session going down", + messages: []*ristypes.RISMessage{ + peerAnnouncement(testPeerIP, testV4Prefix, testV6Prefix), + peerAnnouncement(secondPeerIP, testV4Prefix), + sessionDown(), + }, + want: map[string]map[int][]string{ + testV4Prefix: {testASNB: {secondPeerIP}}, + testV6Prefix: {}, + }, + }, + }) +} + +// TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable hands the +// prefix handler announcements whose AS path gives no origin. One whose path +// holds AS 0 removes the peer's route, as a withdrawal does; one whose path ends +// in an AS set of several members is kept with origin 0. +func TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable(t *testing.T) { + t.Parallel() + + announceWithPath := func(path ristypes.ASPath) *ristypes.RISMessage { + msg := peerAnnouncement(testPeerIP, testV4Prefix) + msg.Path = path + + return msg + } + + checkTableCases(t, []tableCase{ + { + name: "a path whose last AS is 0", + messages: []*ristypes.RISMessage{ + peerAnnouncement(testPeerIP, testV4Prefix), + announceWithPath(ristypes.ASPath{Sequence: []int{testASNA, 0}}), + }, + want: map[string]map[int][]string{testV4Prefix: {}}, + }, + { + name: "a path with AS 0 before its end", + messages: []*ristypes.RISMessage{ + peerAnnouncement(testPeerIP, testV4Prefix), + announceWithPath(decodePath(t, testASNA, 0, testASNB)), + }, + want: map[string]map[int][]string{testV4Prefix: {}}, + }, + { + name: "a path that ends in an AS set", + messages: []*ristypes.RISMessage{ + announceWithPath(decodePath(t, testASNA, []int{testASNC, testASND})), + }, + want: map[string]map[int][]string{testV4Prefix: {0: {testPeerIP}}}, + }, + }) +} + +// checkTableCases hands the messages of each case to a new prefix handler and +// checks its routing table right after, with no flush. +func checkTableCases(t *testing.T, tests []tableCase) { + t.Helper() + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + h := prefixHandlerWithoutTimer(newMockStore()) + for _, msg := range tt.messages { + h.HandleMessage(msg) + } + + for prefix, want := range tt.want { + got := tableOrigins(h.table, prefix) + if !reflect.DeepEqual(got, want) { + t.Errorf("peers of %s by origin AS = %v, want %v", + prefix, got, want) + } + } + }) + } +} + +// tableWritesStore is a real database that hands each write of the routing +// table's changes to written once it is made. +type tableWritesStore struct { + *database.Database + + written chan []routetable.Change +} + +func (s tableWritesStore) WriteRouteChanges( + changes []routetable.Change, newPeers map[string]uint, +) error { + err := s.Database.WriteRouteChanges(changes, newPeers) + s.written <- changes + + return err +} + +// TestPrefixHandlerWritesTheTableChangesEachSecond hands a running prefix +// handler an announcement and then its withdrawal, each far short of a full +// batch. The flush timer must write the change of each to the database. +func TestPrefixHandlerWritesTheTableChangesEachSecond(t *testing.T) { + t.Parallel() + + // The two writes the test waits for, and room for one more. + const writesBuffered = 3 + + store := tableWritesStore{ + Database: newTestDatabase(t), + written: make(chan []routetable.Change, writesBuffered), + } + h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{})) + + defer h.Stop(t.Context()) + + prefix := netip.MustParsePrefix(testV4Prefix) + + for _, step := range []struct { + msg *ristypes.RISMessage + wantRemoved bool + }{ + {msg: peerAnnouncement(testPeerIP, testV4Prefix)}, + {msg: peerWithdrawal(testPeerIP, testV4Prefix), wantRemoved: true}, + } { + h.HandleMessage(step.msg) + + select { + case changes := <-store.written: + if len(changes) != 1 || changes[0].Prefix != prefix || + changes[0].OriginASN != testASNB || + changes[0].Removed != step.wantRemoved { + t.Fatalf("wrote %+v, want the pair of %s and AS%d with "+ + "removed %t", changes, prefix, testASNB, step.wantRemoved) + } + case <-time.After(tableWriteTimeout): + t.Fatalf("the routing table's change was not written within %s", + tableWriteTimeout) + } + } +} + +// errTestTableWrite is the error a test store's failing write of the routing +// table's changes returns. +var errTestTableWrite = errors.New("test: the write of the table's changes failed") + +// firstTableWriteFailsStore is a real database whose first write of the +// routing table's changes fails, writing nothing. writes counts the writes +// asked for. +type firstTableWriteFailsStore struct { + *database.Database + + writes int +} + +func (s *firstTableWriteFailsStore) WriteRouteChanges( + changes []routetable.Change, newPeers map[string]uint, +) error { + s.writes++ + if s.writes == 1 { + return errTestTableWrite + } + + return s.Database.WriteRouteChanges(changes, newPeers) +} + +// TestPrefixHandlerWritesAFailedTableWriteAgain runs two flushes into a +// database whose first write of the routing table's changes fails. Before the +// first, two new peers announce a prefix each; before the second, one of them +// announces its prefix again with another path to the same origin AS. The +// second write must hold the changes and new peers of the first, with the later +// path, so the database holds the table that is in memory. +func TestPrefixHandlerWritesAFailedTableWriteAgain(t *testing.T) { + t.Parallel() + + db := newTestDatabase(t) + store := &firstTableWriteFailsStore{Database: db} + h := prefixHandlerWithoutTimer(store) + + first := peerAnnouncement(testPeerIP, testV4Prefix) + h.HandleMessage(first) + h.HandleMessage(peerAnnouncement(secondPeerIP, otherV4Prefix)) + h.flushBatch() + + laterPath := []int{testASNC, testASNB} + later := peerAnnouncement(testPeerIP, testV4Prefix) + later.Path = ristypes.ASPath{Sequence: laterPath} + later.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter) + h.HandleMessage(later) + h.flushBatch() + + if store.writes != 2 { + t.Fatalf("the store was asked for %d writes of the table's changes, "+ + "want 2", store.writes) + } + + read, err := db.ReadRouteTable(t.Context()) + if err != nil { + t.Fatalf("ReadRouteTable: %v", err) + } + + checkSameTable(t, read, h.table, testV4Prefix, otherV4Prefix) + + origins := read.Origins(netip.MustParsePrefix(testV4Prefix)) + if len(origins) != 1 || len(origins[0].Routes) != 1 || + !slices.Equal(origins[0].Routes[0].ASPath, laterPath) { + t.Errorf("Origins(%s) = %+v, want one route with the path %v", + testV4Prefix, origins, laterPath) + } +} + +// tableWriteSizesStore is a real database that records how many changes and +// how many new peers each write of the routing table's changes holds. +type tableWriteSizesStore struct { + *database.Database + + changes []int + newPeers []int +} + +func (s *tableWriteSizesStore) WriteRouteChanges( + changes []routetable.Change, newPeers map[string]uint, +) error { + s.changes = append(s.changes, len(changes)) + s.newPeers = append(s.newPeers, len(newPeers)) + + return s.Database.WriteRouteChanges(changes, newPeers) +} + +// TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes has one new peer +// change one and a half times as many prefix and origin AS pairs as one write +// takes, then flushes three times. The first write must hold as many pairs as +// one write takes, and the peer. The second must hold the other pairs, and the +// third flush must write nothing. The database must then hold the table in +// memory. +func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) { + t.Parallel() + + const ( + pairs = maxTableChangesPerWrite * 3 / 2 + flushes = 3 + ) + + db := newTestDatabase(t) + store := &tableWriteSizesStore{Database: db} + h := prefixHandlerWithoutTimer(store) + announceManyPrefixes(h.table, pairs) + + for range flushes { + h.flushBatch() + } + + wantChanges := []int{maxTableChangesPerWrite, pairs - maxTableChangesPerWrite} + wantNewPeers := []int{1, 0} + + if !slices.Equal(store.changes, wantChanges) || + !slices.Equal(store.newPeers, wantNewPeers) { + t.Errorf("the writes held %v pairs and %v new peers, want %v and %v", + store.changes, store.newPeers, wantChanges, wantNewPeers) + } + + read, err := db.ReadRouteTable(t.Context()) + if err != nil { + t.Fatalf("ReadRouteTable: %v", err) + } + + checkSameTable(t, read, h.table, firstManyPrefix) +} + +// TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten stops a prefix handler, +// with no deadline, whose routing table holds more changed pairs than two +// writes take. Its last flush writes one write's worth, so Stop must log the +// pairs that no write took. +func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) { + t.Parallel() + + // More than two writes take, so that pairs are left even when the flush + // timer writes once before Stop. + const pairs = 2*maxTableChangesPerWrite + 1 + + var logs bytes.Buffer + + store := &tableWriteSizesStore{Database: newTestDatabase(t)} + table := routetable.New(routetable.Params{}) + announceManyPrefixes(table, pairs) + + h := NewPrefixHandler(store, newErrorLogger(&logs), table) + h.Stop(t.Context()) + + written := 0 + for _, n := range store.changes { + written += n + } + + if logged := lastLogged(t, &logs); logged.TableChanges != pairs-written { + t.Errorf("Stop logged %d table changes not written, want %d", + logged.TableChanges, pairs-written) + } +} + +// readLog holds what readRouteTable logs. +// +//nolint:tagliatelle // field names are the keys readRouteTable logs +type readLog struct { + Msg string `json:"msg"` + Pairs int `json:"pairs"` + Peers int `json:"peers"` + DurationMs *int64 `json:"duration_ms"` +} + +// TestARestartReadsTheRoutingTableBack runs announcements, a withdrawal and a +// peer's session going down through a prefix handler into a database, then +// reads the table back as startup does. The table read must answer as the one +// in memory, and the log must give its pairs and peers. A new handler on the +// table read then takes later updates and writes them as it stops: the +// database must then hold what the first table holds after the same updates. +func TestARestartReadsTheRoutingTableBack(t *testing.T) { + t.Parallel() + + prefixes := []string{testV4Prefix, testV6Prefix, otherV4Prefix, setV4Prefix} + + db := newTestDatabase(t) + h := prefixHandlerWithoutTimer(db) + + viaOtherOrigin := peerAnnouncement(thirdPeerIP, testV4Prefix) + viaOtherOrigin.Path = ristypes.ASPath{Sequence: []int{testASNA, testASNC}} + endsInSet := peerAnnouncement(secondPeerIP, setV4Prefix) + endsInSet.Path = decodePath(t, testASNA, []int{testASNC, testASND}) + + for _, msg := range []*ristypes.RISMessage{ + peerAnnouncement(testPeerIP, testV4Prefix, testV6Prefix), + peerAnnouncement(secondPeerIP, testV4Prefix), + viaOtherOrigin, + endsInSet, + peerAnnouncement(thirdPeerIP, otherV4Prefix), + peerWithdrawal(thirdPeerIP, otherV4Prefix), + peerAnnouncement(fourthPeerIP, otherV4Prefix), + sessionDownOf(fourthPeerIP), + } { + h.HandleMessage(msg) + } + + h.flushBatch() + + var logs bytes.Buffer + + rw := &RouteWatch{ + db: db, + logger: &logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logs, nil))}, + } + + table, err := rw.readRouteTable(t.Context()) + if err != nil { + t.Fatalf("readRouteTable: %v", err) + } + + checkSameTable(t, table, h.table, prefixes...) + + // testV4Prefix with two origin ASes, testV6Prefix and setV4Prefix; four + // peers, the last of which carries nothing. + const wantPairs, wantPeers = 4, 4 + + var logged readLog + + err = json.Unmarshal(bytes.TrimSpace(logs.Bytes()), &logged) + if err != nil || logged.Pairs != wantPairs || logged.Peers != wantPeers || + logged.DurationMs == nil { + t.Errorf("readRouteTable logged %q, want %d pairs, %d peers and the "+ + "duration", logs.String(), wantPairs, wantPeers) + } + + restarted := NewPrefixHandler(db, logger.New(), table) + + for _, msg := range []*ristypes.RISMessage{ + peerWithdrawal(testPeerIP, testV6Prefix), + sessionDownOf(secondPeerIP), + peerAnnouncement(fourthPeerIP, otherV4Prefix), + peerAnnouncement(fifthPeerIP, setV4Prefix), + } { + // The first handler only applies the updates to its table. + h.HandleMessage(msg) + restarted.HandleMessage(msg) + } + + restarted.Stop(t.Context()) + + read, err := db.ReadRouteTable(t.Context()) + if err != nil { + t.Fatalf("ReadRouteTable after the restart: %v", err) + } + + checkSameTable(t, read, h.table, prefixes...) +} + +// peerWithdrawal returns an UPDATE message in which peer withdraws prefixes. +func peerWithdrawal(peer string, prefixes ...string) *ristypes.RISMessage { + return &ristypes.RISMessage{ + Type: ristypes.MessageTypeUpdate, + Peer: peer, + ParsedTimestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC), + Withdrawals: prefixes, + } +} + +// announceManyPrefixes has testPeerIP announce n prefixes in table, with the +// path of peerAnnouncement: the /32 of each address from firstManyPrefix on. +func announceManyPrefixes(table *routetable.Table, n int) { + addr := netip.MustParsePrefix(firstManyPrefix).Addr() + route := routetable.Route{ + Peer: testPeerIP, + ASPath: []int{testASNA, testASNB}, + Time: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC), + } + + for range n { + table.Announce(netip.PrefixFrom(addr, addr.BitLen()), testASNB, route) + addr = addr.Next() + } +} + +// sessionDownOf returns the STATE message that says the BGP session of peer +// went down. +func sessionDownOf(peer string) *ristypes.RISMessage { + return &ristypes.RISMessage{ + Type: ristypes.MessageTypeState, + Peer: peer, + State: ristypes.StateDown, + } +} + +// tableOrigins returns, for each origin AS of prefix in table, the peers that +// carry the prefix with it. +func tableOrigins(table *routetable.Table, prefix string) map[int][]string { + origins := make(map[int][]string) + for _, o := range table.Origins(netip.MustParsePrefix(prefix)) { + origins[o.ASN] = o.Peers + } + + return origins +} + +// checkSameTable checks that got answers as want does for each of prefixes, and +// that the two tables have the same counts. +func checkSameTable( + t *testing.T, got, want *routetable.Table, prefixes ...string, +) { + t.Helper() + + for _, text := range prefixes { + prefix := netip.MustParsePrefix(text) + + gotOrigins, wantOrigins := got.Origins(prefix), want.Origins(prefix) + if !reflect.DeepEqual(gotOrigins, wantOrigins) { + t.Errorf("Origins(%s) = %+v, want %+v", text, gotOrigins, wantOrigins) + } + } + + gotCounts, wantCounts := got.Counts(), want.Counts() + if !reflect.DeepEqual(gotCounts, wantCounts) { + t.Errorf("Counts() = %+v, want %+v", gotCounts, wantCounts) + } +}