diff --git a/README.md b/README.md index d25bdfd..c024249 100644 --- a/README.md +++ b/README.md @@ -145,7 +145,7 @@ and origin AS can have two kept routes, one or none. prefix's traffic to. RouteWatch records the peer itself as the next hop. The page also lists the prefix's origin changes, newest first. An origin change is an AS starting to announce the prefix, when its first peer carries it, or - stopping, when its last peer no longer does. It shows at most the newest + stopping, when its last peer no longer does. The page shows at most the newest 5,000. Each row shows: - the time - announced or withdrawn @@ -577,6 +577,9 @@ What happens at each limit: and after the held-back messages of every peer whose session went down before it. It holds at most 150,000 such messages, about 150 MiB. At that number it takes nothing more off its queue until it has applied one of them. +- Waiting origin changes: the routing table keeps at most 100,000 origin changes + waiting to be written. Past that, it drops the oldest, and the prefix handler + logs how many. The prefix page never shows a dropped origin change. To replace the image's Go soft limit, set `GOMEMLIMIT` in the environment, for example `-e GOMEMLIMIT=3GiB`. A limit close to the routing table's size makes diff --git a/internal/database/originchanges_internal_test.go b/internal/database/originchanges_internal_test.go index 0dd971c..89f950f 100644 --- a/internal/database/originchanges_internal_test.go +++ b/internal/database/originchanges_internal_test.go @@ -58,6 +58,7 @@ func TestOriginChangesPrune(t *testing.T) { db := newRoutesTestDatabase(t) oldPrefix := netip.MustParsePrefix(tablePrefixV4) straddling := netip.MustParsePrefix(tablePrefixShortV4) + busy := netip.MustParsePrefix(tablePrefixBusyV4) other := netip.MustParsePrefix("10.20.0.0/16") // Whole seconds, so the stored times round-trip exactly. @@ -70,6 +71,12 @@ func TestOriginChangesPrune(t *testing.T) { // prune keeps all 200 newer and the newest 300 of the 400 older. changes = append(changes, rows(straddling, now.Add(-100*time.Hour), 400)...) changes = append(changes, rows(straddling, now.Add(-71*time.Hour), 200)...) + // 100 rows older than 72 hours and minOriginChangesPerPrefix+100 newer + // rows for busy. The prune keeps all the newer rows and none of the older, + // because the newer rows alone are more than minOriginChangesPerPrefix. + changes = append(changes, rows(busy, now.Add(-100*time.Hour), 100)...) + changes = append(changes, rows(busy, now.Add(-71*time.Hour), + minOriginChangesPerPrefix+100)...) // One row for other, which the prune leaves alone. changes = append(changes, rows(other, now.Add(-100*time.Hour), 1)...) @@ -83,44 +90,25 @@ func TestOriginChangesPrune(t *testing.T) { t.Fatalf("PruneOriginChanges: %v", err) } - oldTimes := survivorTimes(t, db, oldPrefix, 1000) - if len(oldTimes) != 500 { - t.Errorf("GetOriginChanges(oldPrefix) = %d rows, want 500", len(oldTimes)) - } - // The newest row survives, and the oldest survivor is the 100th oldest, // so a prune that kept the oldest 500 instead would fail. - wantNewest := now.Add(-100 * time.Hour).Add(599 * time.Minute) - if !oldTimes[0].Equal(wantNewest) { - t.Errorf("the newest surviving row's time = %v, want %v", - oldTimes[0], wantNewest) - } - - wantOldest := now.Add(-100 * time.Hour).Add(100 * time.Minute) - if !oldTimes[499].Equal(wantOldest) { - t.Errorf("the oldest surviving row's time = %v, want %v", - oldTimes[499], wantOldest) - } - - straddlingTimes := survivorTimes(t, db, straddling, 1000) - if len(straddlingTimes) != 500 { - t.Errorf("GetOriginChanges(straddling) = %d rows, want 500", - len(straddlingTimes)) - } + checkSurvivors(t, db, oldPrefix, 500, + now.Add(-100*time.Hour).Add(599*time.Minute), + now.Add(-100*time.Hour).Add(100*time.Minute)) // The newest of the newer-than-72-hours rows survives, and the oldest // survivor is the 100th oldest of the older-than-72-hours rows. - wantNewest = now.Add(-71 * time.Hour).Add(199 * time.Minute) - if !straddlingTimes[0].Equal(wantNewest) { - t.Errorf("the newest surviving row's time = %v, want %v", - straddlingTimes[0], wantNewest) - } + checkSurvivors(t, db, straddling, 500, + now.Add(-71*time.Hour).Add(199*time.Minute), + now.Add(-100*time.Hour).Add(100*time.Minute)) - wantOldest = now.Add(-100 * time.Hour).Add(100 * time.Minute) - if !straddlingTimes[499].Equal(wantOldest) { - t.Errorf("the oldest surviving row's time = %v, want %v", - straddlingTimes[499], wantOldest) - } + // Every newer-than-72-hours row survives, so the oldest survivor is the + // first of them and the newest is minOriginChangesPerPrefix+99 minutes + // after it. + checkSurvivors(t, db, busy, minOriginChangesPerPrefix+100, + now.Add(-71*time.Hour). + Add(time.Duration(minOriginChangesPerPrefix+99)*time.Minute), + now.Add(-71*time.Hour)) gotOther, err := db.GetOriginChanges(t.Context(), other, 10) if err != nil { @@ -167,6 +155,33 @@ func survivorTimes( return times } +// checkSurvivors checks that prefix kept exactly wantCount rows, with the +// newest survivor at wantNewest and the oldest at wantOldest. +func checkSurvivors( + t *testing.T, db *Database, prefix netip.Prefix, + wantCount int, wantNewest, wantOldest time.Time, +) { + t.Helper() + + times := survivorTimes(t, db, prefix, wantCount+1) + if len(times) != wantCount { + t.Errorf("GetOriginChanges(%s) = %d rows, want %d", + prefix, len(times), wantCount) + + return + } + + if !times[0].Equal(wantNewest) { + t.Errorf("the newest surviving row's time = %v, want %v", + times[0], wantNewest) + } + + if !times[wantCount-1].Equal(wantOldest) { + t.Errorf("the oldest surviving row's time = %v, want %v", + times[wantCount-1], wantOldest) + } +} + // TestOriginChangesSurviveRouteDeletion checks that deleting a routes row // leaves the prefix's origin_changes rows. func TestOriginChangesSurviveRouteDeletion(t *testing.T) { diff --git a/internal/database/routes_internal_test.go b/internal/database/routes_internal_test.go index 711279f..7b439f6 100644 --- a/internal/database/routes_internal_test.go +++ b/internal/database/routes_internal_test.go @@ -34,6 +34,7 @@ const ( tablePrefixV4 = "10.1.2.0/24" tablePrefixShortV4 = "10.9.0.0/16" + tablePrefixBusyV4 = "10.10.0.0/16" tablePrefixV6 = "2001:db8:1::/48" ) diff --git a/internal/routetable/originchanges.go b/internal/routetable/originchanges.go index 6cd3b34..e1eb41c 100644 --- a/internal/routetable/originchanges.go +++ b/internal/routetable/originchanges.go @@ -6,10 +6,10 @@ import ( "time" ) -// maxWaitingOriginChanges is the most origin changes the table keeps waiting +// MaxWaitingOriginChanges is the most origin changes the table keeps waiting // for the prefix handler to write. The oldest change past it is dropped and // counted. -const maxWaitingOriginChanges = 100000 +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. @@ -76,7 +76,7 @@ func (t *Table) PutBackOriginChanges(changes []OriginChange) { // trimOriginChanges drops the oldest origin changes past the cap and counts // them. The caller holds the lock. func (t *Table) trimOriginChanges() { - if extra := len(t.originChanges) - maxWaitingOriginChanges; extra > 0 { + if extra := len(t.originChanges) - MaxWaitingOriginChanges; extra > 0 { t.originChangesDropped += extra t.originChanges = t.originChanges[extra:] } diff --git a/internal/routetable/originchanges_test.go b/internal/routetable/originchanges_test.go index cd3fa35..857095c 100644 --- a/internal/routetable/originchanges_test.go +++ b/internal/routetable/originchanges_test.go @@ -184,25 +184,24 @@ func TestOriginChangesPutBackPutsRowsInFront(t *testing.T) { 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 { + for range routetable.MaxWaitingOriginChanges + 1 { prefix := netip.PrefixFrom(addr, 32) table.Announce(prefix, asA, routeOf(peerA, asA, 10)) addr = addr.Next() } - changes, dropped := table.TakeOriginChanges(limit) + changes, dropped := table.TakeOriginChanges(routetable.MaxWaitingOriginChanges) if dropped != 1 { t.Errorf("dropped = %d, want 1", dropped) } - if len(changes) != limit { - t.Errorf("len(changes) = %d, want %d", len(changes), limit) + if len(changes) != routetable.MaxWaitingOriginChanges { + t.Errorf("len(changes) = %d, want %d", len(changes), + routetable.MaxWaitingOriginChanges) } } @@ -212,17 +211,15 @@ func TestOriginChangesDropPastTheLimit(t *testing.T) { func TestOriginChangesTakeResetsTheDroppedCount(t *testing.T) { t.Parallel() - const limit = 100000 - table := routetable.New(routetable.Params{}) addr := netip.MustParseAddr("10.0.0.0") - for range limit + 1 { + for range routetable.MaxWaitingOriginChanges + 1 { table.Announce(netip.PrefixFrom(addr, 32), asA, routeOf(peerA, asA, 10)) addr = addr.Next() } - _, dropped := table.TakeOriginChanges(limit) + _, dropped := table.TakeOriginChanges(routetable.MaxWaitingOriginChanges) if dropped != 1 { t.Fatalf("dropped = %d, want 1", dropped) } @@ -241,7 +238,6 @@ func TestOriginChangesPutBackTrimsToTheLimit(t *testing.T) { t.Parallel() const ( - limit = 100000 taken = 20000 added = 20000 ) @@ -263,7 +259,7 @@ func TestOriginChangesPutBackTrimsToTheLimit(t *testing.T) { } } - announce(limit) + announce(routetable.MaxWaitingOriginChanges) takenChanges, _ := table.TakeOriginChanges(taken) @@ -276,8 +272,9 @@ func TestOriginChangesPutBackTrimsToTheLimit(t *testing.T) { t.Errorf("dropped = %d, want %d", dropped, added) } - if len(changes) != limit { - t.Errorf("len(changes) = %d, want %d", len(changes), limit) + if len(changes) != routetable.MaxWaitingOriginChanges { + t.Errorf("len(changes) = %d, want %d", len(changes), + routetable.MaxWaitingOriginChanges) } // The put-back rows are the oldest, so they are the ones dropped. The diff --git a/internal/routewatch/prefixhandler_originchanges_internal_test.go b/internal/routewatch/prefixhandler_originchanges_internal_test.go index a40df80..99ec4af 100644 --- a/internal/routewatch/prefixhandler_originchanges_internal_test.go +++ b/internal/routewatch/prefixhandler_originchanges_internal_test.go @@ -14,6 +14,10 @@ import ( "git.eeqj.de/sneak/routewatch/internal/routetable" ) +// waitingOriginChangesLimit is the routing table's maxWaitingOriginChanges, +// which this package cannot read. +const waitingOriginChangesLimit = 100000 + // TestPrefixHandlerWritesOriginChanges hands a running prefix handler an // announcement and flushes. The flush must write the origin change to the // database. @@ -145,7 +149,7 @@ func TestPrefixHandlerLogsTheDroppedOriginChanges(t *testing.T) { &slog.HandlerOptions{Level: slog.LevelWarn})), } - announceManyPrefixes(h.table, 100001) + announceManyPrefixes(h.table, waitingOriginChangesLimit+1) h.flushBatch() diff --git a/internal/server/prefix_originchanges_internal_test.go b/internal/server/prefix_originchanges_internal_test.go index 75c7eda..2286ca8 100644 --- a/internal/server/prefix_originchanges_internal_test.go +++ b/internal/server/prefix_originchanges_internal_test.go @@ -1,6 +1,7 @@ package server import ( + "fmt" "net/http" "net/http/httptest" "net/netip" @@ -16,6 +17,11 @@ import ( "git.eeqj.de/sneak/routewatch/internal/streamer" ) +// testUncarriedASN is an origin AS the prefix page test's routing table does +// not carry for the prefix, so the AS link the page shows comes from the +// origin changes table. +const testUncarriedASN = 64501 + // newPrefixDetailServer returns a server whose routing table holds testPrefixV4 // with origin AS 64500, and whose database holds the given origin changes. func newPrefixDetailServer( @@ -44,17 +50,18 @@ func newPrefixDetailServer( // TestPrefixDetailShowsOriginChanges checks that the prefix page shows the // origin changes newest first, with the announced and withdrawn rows and the -// AS link. +// AS link. The rows use an AS the routing table does not carry for the +// prefix, so the link comes from the origin changes table. func TestPrefixDetailShowsOriginChanges(t *testing.T) { t.Parallel() prefix := netip.MustParsePrefix(testPrefixV4) s := newPrefixDetailServer(t, []routetable.OriginChange{ - {Prefix: prefix, OriginASN: 64500, Announced: true, PeerCount: 1, - Time: testRouteTime(1)}, - {Prefix: prefix, OriginASN: 64500, Announced: false, PeerCount: 0, - Time: testRouteTime(2)}, + {Prefix: prefix, OriginASN: testUncarriedASN, Announced: true, + PeerCount: 1, Time: testRouteTime(1)}, + {Prefix: prefix, OriginASN: testUncarriedASN, Announced: false, + PeerCount: 0, Time: testRouteTime(2)}, }) rec := httptest.NewRecorder() @@ -79,8 +86,8 @@ func TestPrefixDetailShowsOriginChanges(t *testing.T) { 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") + if !strings.Contains(page, `href="/as/64501"`) { + t.Errorf("page does not link to AS64501") } } @@ -115,15 +122,16 @@ func TestPrefixDetailShowsOriginZeroAsNoKnownOrigin(t *testing.T) { } } -// TestPrefixDetailSaysWhenMoreThan5000Changes checks that the prefix page says -// so when a prefix has more than 5,000 origin changes. -func TestPrefixDetailSaysWhenMoreThan5000Changes(t *testing.T) { +// TestPrefixDetailSaysWhenThereAreMoreChangesThanItShows checks that the +// prefix page says so when a prefix has more origin changes than the page's +// limit. +func TestPrefixDetailSaysWhenThereAreMoreChangesThanItShows(t *testing.T) { t.Parallel() prefix := netip.MustParsePrefix(testPrefixV4) - changes := make([]routetable.OriginChange, 0, 5001) - for i := range 5001 { + changes := make([]routetable.OriginChange, 0, maxOriginChangesOnPage+1) + for i := range maxOriginChangesOnPage + 1 { changes = append(changes, routetable.OriginChange{ Prefix: prefix, OriginASN: 64500, Announced: true, PeerCount: 1, Time: testRouteTime(i), @@ -142,24 +150,27 @@ func TestPrefixDetailSaysWhenMoreThan5000Changes(t *testing.T) { page := rec.Body.String() - if !strings.Contains(page, "more than 5000 changes") { - t.Errorf("page does not say there are more than 5000 changes") + note := fmt.Sprintf("more than %d changes", maxOriginChangesOnPage) + if !strings.Contains(page, note) { + t.Errorf("page does not say there are more than %d changes", + maxOriginChangesOnPage) } - if got := strings.Count(page, "Announced"); got != 5000 { - t.Errorf("page shows %d change rows, want 5000", got) + if got := strings.Count(page, "Announced"); got != maxOriginChangesOnPage { + t.Errorf("page shows %d change rows, want %d", got, + maxOriginChangesOnPage) } } -// TestPrefixDetailShowsNoNoteAtExactly5000Changes checks that a prefix with -// exactly 5,000 origin changes shows all 5,000 and no note. -func TestPrefixDetailShowsNoNoteAtExactly5000Changes(t *testing.T) { +// TestPrefixDetailShowsNoNoteAtTheLimit checks that a prefix with exactly the +// page's limit of origin changes shows all of them and no note. +func TestPrefixDetailShowsNoNoteAtTheLimit(t *testing.T) { t.Parallel() prefix := netip.MustParsePrefix(testPrefixV4) - changes := make([]routetable.OriginChange, 0, 5000) - for i := range 5000 { + changes := make([]routetable.OriginChange, 0, maxOriginChangesOnPage) + for i := range maxOriginChangesOnPage { changes = append(changes, routetable.OriginChange{ Prefix: prefix, OriginASN: 64500, Announced: true, PeerCount: 1, Time: testRouteTime(i), @@ -179,11 +190,13 @@ func TestPrefixDetailShowsNoNoteAtExactly5000Changes(t *testing.T) { page := rec.Body.String() if strings.Contains(page, "more than") { - t.Errorf("page says there are more changes, but a prefix has 5000") + t.Errorf("page says there are more changes, but a prefix has %d", + maxOriginChangesOnPage) } - if got := strings.Count(page, "Announced"); got != 5000 { - t.Errorf("page shows %d change rows, want 5000", got) + if got := strings.Count(page, "Announced"); got != maxOriginChangesOnPage { + t.Errorf("page shows %d change rows, want %d", got, + maxOriginChangesOnPage) } }