From 31aa3a4563f11541ebd80076ebb93eac9837d0c2 Mon Sep 17 00:00:00 2001 From: Rayhaan Jaufeerally Date: Fri, 2 Oct 2026 00:31:28 +0200 Subject: [PATCH] Use the sharded table in the server: sessions choose best routes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The server now keeps what it learns in table::RibTable. The Adj-RIB-In actor is gone, and with it the queue every UPDATE from every session went through. - A session takes a number in the table when it is established and writes its UPDATEs into the table itself. Its routes are removed when it ends, on a blocking thread since that looks at every prefix; a session torn down without cleaning up still has them removed. - For every prefix the best route is chosen (RFC 4271 §9.1.2.2) as routes come and go. Nothing acts on the choice yet beyond showing it. - The UPDATE decoder hands over the bytes of the attributes it kept, so the table finds the attribute set from them without writing the attributes out again. On a session with two-octet AS numbers the bytes are not in the table's form, and the parsed attributes are used. - /api/routes/received marks the best route of each prefix and returns at most `limit` routes (default 1000): the table can hold hundreds of millions. /api/status has the totals, and the status page says how many it is showing. The time a route was received is no longer kept or shown. - The event stream reports changes of best route (best_changed, best_withdrawn) in place of every route received. Each client follows the table's journal with a reader of its own, from the moment it connects: table::RibTable::reader_from_now. - Metrics: bgp_rib_routes, bgp_rib_prefixes, bgp_rib_attribute_sets and bgp_rib_journal_entries replace bgp_adj_rib_in_routes. - A line is no longer logged at INFO for every UPDATE sent or received; a full table is a million of them. They are at DEBUG. - A table with no room for another peer ends the session with Cease / Out of Resources. Four peers sending a synthetic full table each over TCP, 4.9 million routes: 0.80 s, against 4.76 s before. One peer: 0.45 s against 0.69 s. Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 1 + bin/bgp/src/main.rs | 2 +- crates/bench/benches/full_table.rs | 382 ++++++------------ crates/bench/benches/real_table.rs | 126 +----- crates/integration-tests/src/test_server.rs | 100 ++++- crates/integration-tests/tests/announce.rs | 12 +- .../tests/basic_connection.rs | 35 +- crates/integration-tests/tests/best_route.rs | 239 +++++++++++ crates/integration-tests/tests/collision.rs | 4 +- crates/integration-tests/tests/failure.rs | 68 ++-- crates/integration-tests/tests/http_api.rs | 2 +- crates/integration-tests/tests/interop.rs | 2 +- crates/integration-tests/tests/negotiation.rs | 7 +- .../integration-tests/tests/update_errors.rs | 17 +- crates/packet/src/message.rs | 27 +- crates/packet/src/update_decode.rs | 78 +++- crates/packet/tests/properties.rs | 15 +- crates/rib/src/adj_rib_in.rs | 277 ------------- crates/rib/src/entry.rs | 197 --------- crates/rib/src/error.rs | 6 +- crates/rib/src/event.rs | 16 - crates/rib/src/lib.rs | 11 +- crates/rib/src/local_rib.rs | 8 +- crates/rib/src/peer.rs | 14 + crates/rib/src/table/attrs.rs | 18 + crates/rib/src/table/journal.rs | 24 +- crates/rib/src/table/journal_tests.rs | 25 ++ crates/rib/src/table/mod.rs | 23 ++ crates/server/src/connection.rs | 219 ++++++---- crates/server/src/error.rs | 8 +- crates/server/src/fsm.rs | 19 +- crates/server/src/lib.rs | 1 + crates/server/src/metrics.rs | 35 +- crates/server/src/routes.rs | 62 +++ crates/server/src/server.rs | 40 +- crates/server/src/session.rs | 14 +- crates/web/Cargo.toml | 1 + crates/web/src/api.rs | 96 +++-- crates/web/src/events.rs | 137 +++++-- crates/web/src/lib.rs | 6 +- crates/web/src/metrics.rs | 24 +- crates/web/static/statusz.html | 17 +- fuzz/src/lib.rs | 10 +- 43 files changed, 1213 insertions(+), 1212 deletions(-) create mode 100644 crates/integration-tests/tests/best_route.rs delete mode 100644 crates/rib/src/adj_rib_in.rs delete mode 100644 crates/rib/src/entry.rs delete mode 100644 crates/rib/src/event.rs create mode 100644 crates/rib/src/peer.rs create mode 100644 crates/server/src/routes.rs diff --git a/Cargo.lock b/Cargo.lock index 3dbfa6b..97a412e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -274,6 +274,7 @@ version = "0.0.2" dependencies = [ "axum", "bgp-packet", + "bgp-primitives", "bgp-rib", "bgp-server", "futures", diff --git a/bin/bgp/src/main.rs b/bin/bgp/src/main.rs index 5786a44..071934c 100644 --- a/bin/bgp/src/main.rs +++ b/bin/bgp/src/main.rs @@ -45,7 +45,7 @@ async fn main() -> Result<()> { ), registry: server.registry.clone(), local_rib: server.local_rib.clone(), - adj_rib_in: server.adj_rib_in.clone(), + rib: Arc::clone(&server.rib), }; tokio::spawn(async move { if let Err(e) = bgp_web::serve(state, http_endpoint).await { diff --git a/crates/bench/benches/full_table.rs b/crates/bench/benches/full_table.rs index 64b9dce..4c55887 100644 --- a/crates/bench/benches/full_table.rs +++ b/crates/bench/benches/full_table.rs @@ -1,25 +1,28 @@ -//! What it costs to take in the Internet routing table. +//! What it costs to take in the Internet routing table, from the bytes on +//! the wire to a chosen best route. //! -//! Run with `cargo bench -p bgp-bench`. Set `BENCH_PEERS` (default 4) and -//! `BENCH_SCALE_DOWN` (default 1; 10 makes the table ten times smaller) to -//! change the size. +//! Run with `cargo bench -p bgp-bench --bench full_table`. Set +//! `BENCH_PEERS` (default 4) and `BENCH_SCALE_DOWN` (default 1; 10 makes the +//! table ten times smaller) to change the size. The table is synthetic (see +//! the crate documentation); `real_table` runs on a real one. //! -//! This is a stopwatch and an allocation counter around whole operations at -//! full size, not a statistical micro-benchmark: the questions are "how -//! long until a full table is in" and "how much memory does it take", and -//! their answers are seconds and gigabytes. +//! This is a stopwatch around whole operations at full size, not a +//! statistical micro-benchmark: the questions are "how long until a full +//! table is in" and "how much memory does it take", and their answers are +//! seconds and gigabytes. use std::hint::black_box; -use std::net::{Ipv4Addr, SocketAddr}; +use std::net::Ipv4Addr; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{Duration, Instant}; -use bgp_bench::memory::Region; +use bgp_bench::memory::{Region, describe}; use bgp_bench::{Prefixes, Shape, announcements, encoded, wire_context}; use bgp_integration_tests::test_peer::TestPeer; use bgp_integration_tests::test_server::TestServer; -use bgp_packet::message::{Message, PathAttribute, UpdateMessage}; +use bgp_packet::message::{Message, PathAttribute}; use bgp_primitives::ip_prefix::IpPrefix; -use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate, RibEntry, SnapshotEntry}; +use bgp_rib::table::{PeerIndex, PeerInfo, RibTable}; use bytes::BytesMut; fn env(name: &str, default: usize) -> usize { @@ -29,13 +32,7 @@ fn env(name: &str, default: usize) -> usize { .unwrap_or(default) } -/// Bytes allocated and not freed since `region` began; zero when memory -/// is not being counted (see `bgp_bench::memory`). -fn retained(region: &Region) -> i64 { - region.held().unwrap_or(0) -} - -fn mib(bytes: i64) -> f64 { +fn mib(bytes: usize) -> f64 { bytes as f64 / (1024.0 * 1024.0) } @@ -52,66 +49,39 @@ fn row(what: &str, elapsed: Duration, detail: String) { println!("{what:<44} {:>9.3} s {detail}", elapsed.as_secs_f64()); } -fn peer_id(peer: u32) -> PeerId { - PeerId { - addr: SocketAddr::new(Ipv4Addr::new(192, 0, 2, peer as u8).into(), 179), - asn: 64512 + peer, - } -} - fn main() { - if Region::begin().held().is_none() { - println!( - "Memory is not counted in this run: sizes print as 0. For sizes (and slower\ntimings) add --features count-memory.\n" - ); - } let peers = env("BENCH_PEERS", 4) as u32; let shape = Shape::INTERNET.scaled_down(env("BENCH_SCALE_DOWN", 1)); - - println!("== Sizes of the types a route is made of"); - for (name, size) in [ - ("IpPrefix", size_of::()), - ("RibEntry", size_of::()), - ( - "(IpPrefix, RibEntry), a table slot", - size_of::<(IpPrefix, RibEntry)>(), - ), - ("PathAttribute", size_of::()), - ("SnapshotEntry", size_of::()), - ] { - println!("{name:<44} {size:>5} bytes"); - } + let threads = std::thread::available_parallelism().map_or(1, |n| n.get()); println!( - "\n== Table: {} IPv4 + {} IPv6 prefixes, from each of {peers} peers", + "== Table: {} IPv4 + {} IPv6 prefixes, from each of {peers} peers", shape.ipv4_prefixes, shape.ipv6_prefixes ); + println!( + "(IpPrefix is {} bytes, PathAttribute {})", + size_of::(), + size_of::() + ); let prefixes = Prefixes::generate(shape); - let per_peer: Vec> = (1..=peers) - .map(|peer| announcements(&prefixes, shape, peer)) + let wire: Vec> = (1..=peers) + .map(|peer| encoded(&announcements(&prefixes, shape, peer))) .collect(); - let wire: Vec> = per_peer.iter().map(|updates| encoded(updates)).collect(); - let updates: usize = per_peer.iter().map(Vec::len).sum(); + let updates: usize = wire.iter().map(Vec::len).sum(); let routes = shape.prefixes() * peers as usize; let wire_bytes: usize = wire.iter().flatten().map(BytesMut::len).sum(); println!( "{updates} UPDATE messages, {:.1} prefixes each, {:.1} MiB on the wire, {routes} routes", routes as f64 / updates as f64, - mib(wire_bytes as i64), + mib(wire_bytes), ); println!("\n== Reading the messages (one thread)"); let ctx = wire_context(); let started = Instant::now(); - let decoded: Vec> = wire - .iter() - .map(|messages| { - messages - .iter() - .map(|m| Message::decode(&ctx, m).expect("what we wrote is well formed")) - .collect() - }) - .collect(); + for message in wire.iter().flatten() { + black_box(Message::decode(&ctx, message).expect("what we wrote is well formed")); + } let elapsed = started.elapsed(); row( "decode UPDATEs", @@ -120,224 +90,126 @@ fn main() { "{} routes, {} messages, {:.0} MiB/s", per_second(routes, elapsed), per_second(updates, elapsed), - mib(wire_bytes as i64) / elapsed.as_secs_f64() - ), - ); - drop(decoded); - - floor(&prefixes, &per_peer, routes); - - println!("\n== The same, as the daemon does it today (one thread)"); - // From here to the filled table: what the routes cost to hold. - let table_region = Region::begin(); - let region = Region::begin(); - let started = Instant::now(); - let received: Vec> = per_peer - .iter() - .map(|updates| updates.iter().map(ReceivedUpdate::from_message).collect()) - .collect(); - let elapsed = started.elapsed(); - row( - "UPDATE -> routes (ReceivedUpdate)", - elapsed, - format!( - "{} routes, holding {:.0} MiB", - per_second(routes, elapsed), - mib(retained(®ion)) + mib(wire_bytes) / elapsed.as_secs_f64() ), ); + println!("\n== From bytes to best routes, without the network"); + for threads in [1, threads] { + into_table(&wire, routes, threads); + } + + println!("\n== End to end: {peers} peers send the table over TCP at once"); let runtime = tokio::runtime::Builder::new_multi_thread() .enable_all() .build() .expect("a runtime"); - - println!("\n== Adj-RIB-In (the actor, fed from one task)"); runtime.block_on(async { - let rib = AdjRibIn::spawn(); - let started = Instant::now(); - for (peer, updates) in received.into_iter().enumerate() { - let peer = peer_id(peer as u32 + 1); - for update in updates { - rib.apply(peer, update).await.expect("the actor runs"); - } + let server = TestServer::start(65000).await.expect("a server"); + let mut senders = Vec::new(); + for (index, messages) in wire.iter().enumerate() { + let peer = index as u32 + 1; + let mut session = TestPeer::connect(server.addr).await.expect("a connection"); + session + .handshake(64512 + peer, Ipv4Addr::new(10, 0, 0, peer as u8), 90) + .await + .expect("a session"); + let stream: Vec = messages.iter().flat_map(|m| m.iter().copied()).collect(); + senders.push((session, stream)); } - let stored = rib.len().await.expect("the actor runs"); - let elapsed = started.elapsed(); - assert_eq!(stored, routes); - let held = retained(&table_region); - row( - "insert every route", - elapsed, - format!( - "{} routes; table holds {:.0} MiB = {:.0} bytes a route", - per_second(routes, elapsed), - mib(held), - held as f64 / routes as f64 - ), - ); - let started = Instant::now(); - black_box(rib.len().await.expect("the actor runs")); - row("count routes", started.elapsed(), String::new()); - - let before = Region::begin(); - let started = Instant::now(); - let snapshot = rib.snapshot().await.expect("the actor runs"); - let elapsed = started.elapsed(); - row( - "snapshot (what /api/routes/received does)", - elapsed, - format!( - "copies {:.0} MiB; nothing else is applied meanwhile", - mib(retained(&before)) - ), - ); - drop(snapshot); - - // A second announcement of everything by one peer: every route is - // replaced, as after a route refresh. - let again: Vec = per_peer[0] - .iter() - .map(ReceivedUpdate::from_message) - .collect(); - let started = Instant::now(); - for update in again { - rib.apply(peer_id(1), update).await.expect("the actor runs"); - } - rib.len().await.expect("the actor runs"); - let elapsed = started.elapsed(); - row( - "replace one peer's routes", - elapsed, - per_second(shape.prefixes(), elapsed) + " routes", - ); - - let started = Instant::now(); - let removed = rib.remove_peer(peer_id(1)).await.expect("the actor runs"); - let elapsed = started.elapsed(); - assert_eq!(removed, shape.prefixes()); - row( - "remove one peer (session down)", - elapsed, - per_second(removed, elapsed) + " routes", - ); - }); - - println!("\n== End to end: {peers} peers send the table over TCP at once"); - for logging in [false, true] { - // The daemon logs a line for every UPDATE at INFO. With no subscriber - // installed that costs almost nothing; this shows what it costs - // when the lines are formatted, as they are in production (here to - // nowhere, so without the cost of the terminal or the disk). - if logging { - tracing::subscriber::set_global_default( - tracing_subscriber::fmt() - .with_writer(std::io::sink) - .with_max_level(tracing::Level::INFO) - .finish(), - ) - .expect("no subscriber was installed before"); - } - runtime.block_on(async { - let server = TestServer::start(65000).await.expect("a server"); - let mut senders = Vec::new(); - for (index, messages) in wire.iter().enumerate() { - let peer = index as u32 + 1; - let mut session = TestPeer::connect(server.addr).await.expect("a connection"); - session - .handshake(64512 + peer, Ipv4Addr::new(10, 0, 0, peer as u8), 90) - .await - .expect("a session"); - let stream: Vec = messages.iter().flat_map(|m| m.iter().copied()).collect(); - senders.push((session, stream)); - } - let started = Instant::now(); - let sending: Vec<_> = senders - .into_iter() - .map(|(mut session, stream)| { - tokio::spawn(async move { - session.send_raw(&stream).await.expect("the server reads"); - // Held open until the table is in. - session - }) + let sending: Vec<_> = senders + .into_iter() + .map(|(mut session, stream)| { + tokio::spawn(async move { + session.send_raw(&stream).await.expect("the server reads"); + // Held open until the table is in. + session }) - .collect(); - while server.adj_rib_in.len().await.expect("the actor runs") < routes { - tokio::time::sleep(Duration::from_millis(5)).await; - } - let elapsed = started.elapsed(); - row( - if logging { - "with a log line per UPDATE" - } else { - "with logging off" - }, - elapsed, - per_second(routes, elapsed) + " routes", - ); - for session in sending { - drop(session.await); - } - }); - } + }) + .collect(); + while server.rib.len() < routes { + tokio::time::sleep(Duration::from_millis(2)).await; + } + let elapsed = started.elapsed(); + row( + "sessions, state machine, table", + elapsed, + format!( + "{} routes; {} prefixes with a best route", + per_second(routes, elapsed), + server.rib.prefix_count() + ), + ); + for session in sending { + drop(session.await); + } + }); } -/// A straw man, not a design: the same routes in the plainest compact form, -/// to show how far the current table is from what a hash table can do. -/// -/// A prefix is one integer and a route is the number of its attribute set -/// (here simply the number of the UPDATE that announced it; a real table -/// would intern the sets). One map per peer and address family, filled on -/// one thread with no channel in between. -fn floor(prefixes: &Prefixes, per_peer: &[Vec], routes: usize) { - use std::collections::HashMap; - - fn key(prefix: &IpPrefix) -> u128 { - let mut address = [0u8; 16]; - address[..prefix.bytes().len()].copy_from_slice(prefix.bytes()); - // The address in the high bits, the length in the low eight. An IPv6 - // prefix longer than /120 would not fit; the table has none. - (u128::from_be_bytes(address) & !0xff) | u128::from(prefix.length()) - } - - println!("\n== Straw man: packed prefix -> attribute set number, no actor"); +/// What a session does with each message, without the session: decode it, +/// find its attribute set, put its routes in the table, choose the best. +/// `threads` threads each take whole peers. +fn into_table(wire: &[Vec], routes: usize, threads: usize) { let region = Region::begin(); + let table = RibTable::new(threads * 16); + let peers: Vec = (0..wire.len()) + .map(|peer| { + table + .add_peer(PeerInfo { + asn: 64512 + peer as u32 + 1, + router_id: Ipv4Addr::from(peer as u32 + 1), + address: Ipv4Addr::new(192, 0, 2, peer as u8 + 1).into(), + ibgp: false, + }) + .expect("room for the peer") + }) + .collect(); + let ctx = wire_context(); + let next = AtomicUsize::new(0); let started = Instant::now(); - let mut tables: Vec<(HashMap, HashMap)> = Vec::new(); - for updates in per_peer { - let mut v4: HashMap = HashMap::with_capacity(prefixes.ipv4.len()); - let mut v6: HashMap = HashMap::with_capacity(prefixes.ipv6.len()); - for (number, update) in updates.iter().enumerate() { - for prefix in &update.announced_routes { - v4.insert( - (key(prefix) >> 64) as u64 | u64::from(prefix.length()), - number as u32, - ); - } - for attribute in &update.path_attributes { - if let PathAttribute::MpReachNlri(mp) = attribute { - for prefix in &mp.prefixes { - v6.insert(key(prefix), number as u32); + std::thread::scope(|scope| { + for _ in 0..threads { + scope.spawn(|| { + loop { + let at = next.fetch_add(1, Ordering::Relaxed); + let Some(messages) = wire.get(at) else { + break; + }; + for message in messages { + let decoded = + Message::decode_with_report(&ctx, message).expect("well formed"); + let Message::Update(update) = decoded.message else { + continue; + }; + let wire = decoded.attributes_wire.as_deref().expect("kept"); + if !update.announced_routes.is_empty() { + let attrs = table.attrs().intern_received(wire, None).expect("valid"); + table.announce(peers[at], &attrs, &update.announced_routes); + } + for attribute in &update.path_attributes { + if let PathAttribute::MpReachNlri(mp) = attribute { + let attrs = table + .attrs() + .intern_received(wire, Some(&mp.next_hop)) + .expect("valid"); + table.announce(peers[at], &attrs, &mp.prefixes); + } + } } } - } + }); } - tables.push((v4, v6)); - } + }); let elapsed = started.elapsed(); - let stored: usize = tables.iter().map(|(v4, v6)| v4.len() + v6.len()).sum(); - assert_eq!(stored, routes); - let held = retained(®ion); + assert_eq!(table.len(), routes); row( - "insert every route", + &format!("decode, intern, insert, choose: {threads} thread(s)"), elapsed, format!( - "{} routes; {:.0} MiB = {:.0} bytes a route (attribute sets not counted)", + "{} routes; {}", per_second(routes, elapsed), - mib(held), - held as f64 / routes as f64 + describe(region.held(), routes, "route") ), ); - black_box(tables); } diff --git a/crates/bench/benches/real_table.rs b/crates/bench/benches/real_table.rs index fba8768..c357cd8 100644 --- a/crates/bench/benches/real_table.rs +++ b/crates/bench/benches/real_table.rs @@ -7,18 +7,17 @@ //! the peers with the most routes (default: all of them). use std::collections::HashSet; -use std::net::{Ipv4Addr, SocketAddr}; +use std::net::Ipv4Addr; use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::time::{Duration, Instant, SystemTime}; +use std::time::{Duration, Instant}; use bgp_bench::memory::Region; use bgp_bench::mrt::{Dump, parse_attributes}; use bgp_packet::message::{NlriNextHop, PathAttribute}; use bgp_primitives::ip_prefix::IpPrefix; use bgp_rib::table::{Event, PeerIndex, PeerInfo, RibTable}; -use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate, RibEntry}; /// Bytes allocated and not freed since `region` began; zero when memory /// is not being counted (see `bgp_bench::memory`). @@ -105,13 +104,6 @@ fn main() { println!("{unparsed} attribute sets could not be parsed; their routes are left out"); } - let runtime = tokio::runtime::Builder::new_multi_thread() - .enable_all() - .build() - .expect("a runtime"); - current_table(&runtime, &dump, &peers, &parsed); - drop(runtime); - let threads = std::thread::available_parallelism().map_or(1, |n| n.get()); sharded_table(&dump, &peers, &parsed, 1); sharded_table(&dump, &peers, &parsed, threads); @@ -169,120 +161,6 @@ fn describe(dump: &Dump, peers: &[usize]) { ); } -fn peer_id(dump: &Dump, peer: usize) -> PeerId { - PeerId { - // The port tells apart peers that share an address in the dump. - addr: SocketAddr::new(dump.peers[peer].address, peer as u16), - asn: dump.peers[peer].asn, - } -} - -/// A peer's routes as the UPDATEs that would have announced them: routes -/// that share their attributes arrive together. -fn as_updates( - dump: &Dump, - peer: usize, - parsed: &[Option], - received_at: SystemTime, -) -> Vec { - let mut routes = dump.routes[peer].clone(); - routes.sort_by_key(|(_, set)| *set); - routes - .chunk_by(|a, b| a.1 == b.1) - .filter_map(|group| { - let (attributes, next_hop) = parsed[group[0].1 as usize].as_ref()?; - Some(ReceivedUpdate { - withdrawn: vec![], - announced: group - .iter() - .map(|(prefix, _)| { - ( - *prefix, - RibEntry { - path_attributes: Arc::clone(attributes), - next_hop: next_hop.clone(), - received_at, - }, - ) - }) - .collect(), - }) - }) - .collect() -} - -fn current_table( - runtime: &tokio::runtime::Runtime, - dump: &Dump, - peers: &[usize], - parsed: &[Option], -) { - println!("\n== Adj-RIB-In as it is today (the actor, fed from one task)"); - // From before the routes exist to the filled table: what it holds, - // apart from the attribute sets, which were parsed above. - let region = Region::begin(); - let now = SystemTime::now(); - let updates: Vec<(PeerId, Vec)> = peers - .iter() - .map(|&peer| (peer_id(dump, peer), as_updates(dump, peer, parsed, now))) - .collect(); - let routes: usize = updates - .iter() - .flat_map(|(_, updates)| updates) - .map(|update| update.announced.len()) - .sum(); - let messages: usize = updates.iter().map(|(_, updates)| updates.len()).sum(); - println!( - "{routes} routes in {messages} UPDATEs, {:.1} routes each", - routes as f64 / messages as f64 - ); - let first = updates[0].0; - let first_routes: usize = updates[0].1.iter().map(|u| u.announced.len()).sum(); - - runtime.block_on(async { - let rib = AdjRibIn::spawn(); - let started = Instant::now(); - for (peer, updates) in updates { - for update in updates { - rib.apply(peer, update).await.expect("the actor runs"); - } - } - let stored = rib.len().await.expect("the actor runs"); - let elapsed = started.elapsed(); - let held = retained(®ion); - row( - "insert every route", - elapsed, - format!( - "{} routes; {stored} stored; {:.0} MiB = {:.0} bytes a route, attribute sets not counted", - per_second(routes, elapsed), - mib(held), - held as f64 / stored as f64 - ), - ); - - let before = Region::begin(); - let started = Instant::now(); - let snapshot = rib.snapshot().await.expect("the actor runs"); - row( - "snapshot", - started.elapsed(), - format!("copies {:.0} MiB", mib(retained(&before))), - ); - drop(snapshot); - - let started = Instant::now(); - let removed = rib.remove_peer(first).await.expect("the actor runs"); - let elapsed = started.elapsed(); - assert_eq!(removed, first_routes); - row( - "remove one peer (session down)", - elapsed, - per_second(removed, elapsed) + " routes", - ); - }); -} - /// The sharded table, filled by `threads` threads. Each takes whole peers, /// as sessions would, and for every UPDATE interns the attributes and then /// announces the prefixes: everything between a decoded message and a diff --git a/crates/integration-tests/src/test_server.rs b/crates/integration-tests/src/test_server.rs index c4cad15..fa01299 100644 --- a/crates/integration-tests/src/test_server.rs +++ b/crates/integration-tests/src/test_server.rs @@ -1,6 +1,10 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::sync::Arc; -use bgp_rib::{AdjRibIn, LocalRib}; +use bgp_packet::message::{NlriNextHop, PathAttribute}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::LocalRib; +use bgp_rib::table::{Event, PeerIndex, Reader, RibTable}; use bgp_server::config::Config; use bgp_server::error::ServerError; use bgp_server::server::BgpServerBuilder; @@ -16,8 +20,8 @@ pub struct TestServer { handle: JoinHandle>, /// The single IPv4 loopback address the server is bound to. pub addr: SocketAddr, - /// Handle to the server's Adj-RIB-In for test introspection. - pub adj_rib_in: AdjRibIn, + /// The server's routing table. + pub rib: Arc, /// Handle to the server's locally originated routes. pub local_rib: LocalRib, /// Live view of the server's established sessions. @@ -58,7 +62,7 @@ impl TestServer { .into_iter() .next() .expect("at least one listener"); - let adj_rib_in = bound.adj_rib_in.clone(); + let rib = Arc::clone(&bound.rib); let local_rib = bound.local_rib.clone(); let registry = bound.registry.clone(); @@ -66,7 +70,7 @@ impl TestServer { Ok(Self { handle, addr, - adj_rib_in, + rib, local_rib, registry, local_asn, @@ -74,7 +78,91 @@ impl TestServer { } } +/// A route in the server's table, as a test wants to look at it. +#[derive(Debug, Clone)] +pub struct ReceivedRoute { + pub prefix: IpPrefix, + pub peer: PeerIndex, + /// The route's attributes, without MP_REACH_NLRI. + pub attributes: Vec, + /// The next hop, if the route came in MP_REACH_NLRI. + pub next_hop: Option, + /// Whether it is the best route for its prefix. + pub best: bool, +} + +/// A change of a prefix's best route. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RouteChange { + /// The prefix has a best route, or another one than before. + Announced { prefix: IpPrefix, peer: PeerIndex }, + /// The prefix has no route any more. + Withdrawn { prefix: IpPrefix }, +} + +/// The changes of best route in a server's table from the moment this was +/// made, read from the table's journal. +pub struct RouteChanges { + reader: Reader, + pending: std::collections::VecDeque, +} + +impl RouteChanges { + /// The next change, waiting for one if need be. Changes to one prefix + /// that happen before they are read arrive as one. + pub async fn recv(&mut self) -> Result { + loop { + if let Some(change) = self.pending.pop_front() { + return Ok(change); + } + self.reader.changed().await; + self.read(); + } + } + + fn read(&mut self) { + let pending = &mut self.pending; + let mut read = |event: Event<'_>| match event { + Event::Best { prefix, peer, .. } => { + pending.push_back(RouteChange::Announced { prefix, peer }) + } + Event::Withdrawn { prefix } => pending.push_back(RouteChange::Withdrawn { prefix }), + Event::Reset | Event::Synced => {} + }; + while !self.reader.poll(1024, &mut read).caught_up {} + } +} + impl TestServer { + /// Every route in the server's table. + pub fn received_routes(&self) -> Vec { + let mut routes = Vec::new(); + self.rib.for_each_route(|prefix, peer, attrs, best| { + let (attributes, next_hop) = attrs.attributes(); + routes.push(ReceivedRoute { + prefix: *prefix, + peer, + attributes, + next_hop, + best, + }); + }); + routes + } + + /// Follows the changes of best route from now on. + pub fn route_changes(&self) -> RouteChanges { + let mut changes = RouteChanges { + reader: self.rib.reader(), + pending: Default::default(), + }; + // A new reader is first given the table as it is. That is not a + // change. + changes.read(); + changes.pending.clear(); + changes + } + /// Waits for the server to stop by itself and returns why. A healthy /// server never does. pub async fn stopped(&mut self) -> Result<(), ServerError> { @@ -96,7 +184,7 @@ impl TestServer { ), registry: self.registry.clone(), local_rib: self.local_rib.clone(), - adj_rib_in: self.adj_rib_in.clone(), + rib: Arc::clone(&self.rib), }; let handle = tokio::spawn(bgp_web::serve_on(state, listener)); Ok((base_url, HttpGuard { handle })) diff --git a/crates/integration-tests/tests/announce.rs b/crates/integration-tests/tests/announce.rs index 3685c00..4a46498 100644 --- a/crates/integration-tests/tests/announce.rs +++ b/crates/integration-tests/tests/announce.rs @@ -5,7 +5,7 @@ use std::net::Ipv4Addr; use std::time::Duration; use bgp_integration_tests::test_peer::TestPeer; -use bgp_integration_tests::test_server::TestServer; +use bgp_integration_tests::test_server::{RouteChange, TestServer}; use bgp_packet::message::{ AsPathAttribute, CommunitiesPathAttribute, LargeCommunitiesPathAttribute, LargeCommunity, NextHopPathAttribute, OriginPathAttribute, PathAttribute, UpdateMessage, @@ -285,7 +285,7 @@ async fn test_keepalives_continue_alongside_updates() -> Result<()> { #[tokio::test] async fn test_received_routes_removed_when_session_closes() -> Result<()> { let server = TestServer::start(LOCAL_ASN).await?; - let mut rib_events = server.adj_rib_in.subscribe(); + let mut rib_events = server.route_changes(); let mut session_events = server.registry.subscribe(); let mut peer = established_peer(&server).await?; wait_for_sessions(&server, 1).await?; @@ -304,22 +304,22 @@ async fn test_received_routes_removed_when_session_closes() -> Result<()> { .await?; for _ in 0..2 { let ev = tokio::time::timeout(TIMEOUT, rib_events.recv()).await??; - assert!(matches!(ev, bgp_rib::RibEvent::Announced { .. }), "{ev:?}"); + assert!(matches!(ev, RouteChange::Announced { .. }), "{ev:?}"); } - assert_eq!(server.adj_rib_in.snapshot().await.unwrap().len(), 2); + assert_eq!(server.received_routes().len(), 2); drop(peer); let mut withdrawn = vec![]; for _ in 0..2 { match tokio::time::timeout(TIMEOUT, rib_events.recv()).await?? { - bgp_rib::RibEvent::Withdrawn { prefix, .. } => withdrawn.push(prefix.to_string()), + RouteChange::Withdrawn { prefix, .. } => withdrawn.push(prefix.to_string()), other => panic!("expected Withdrawn, got {other:?}"), } } withdrawn.sort(); assert_eq!(withdrawn, vec!["198.51.100.0/24", "203.0.113.0/24"]); - assert!(server.adj_rib_in.snapshot().await.unwrap().is_empty()); + assert!(server.received_routes().is_empty()); // Withdrawals precede the Down event. let _up = tokio::time::timeout(TIMEOUT, session_events.recv()).await??; diff --git a/crates/integration-tests/tests/basic_connection.rs b/crates/integration-tests/tests/basic_connection.rs index aafe502..1ad0d53 100644 --- a/crates/integration-tests/tests/basic_connection.rs +++ b/crates/integration-tests/tests/basic_connection.rs @@ -2,6 +2,7 @@ use std::net::Ipv4Addr; use std::time::Duration; use bgp_integration_tests::test_peer::TestPeer; +use bgp_integration_tests::test_server::RouteChange; use bgp_integration_tests::test_server::TestServer; use bgp_packet::errors::{BgpError, CeaseSubcode, FsmSubcode, MsgHeaderSubcode, UpdateMsgSubcode}; use bgp_packet::message::{ @@ -11,7 +12,6 @@ use bgp_packet::message::{ }; use bgp_primitives::constants::{AddressFamilyId, Afi, Safi}; use bgp_primitives::ip_prefix::IpPrefix; -use bgp_rib::RibEvent; use eyre::Result; const TIMEOUT: Duration = Duration::from_secs(5); @@ -229,14 +229,7 @@ async fn test_message_split_across_a_keepalive_tick_is_still_one_message() -> Re peer.send_raw(&update[10..]).await?; tokio::time::timeout(TIMEOUT, async { - while !server - .adj_rib_in - .snapshot() - .await - .unwrap() - .iter() - .any(|e| e.prefix == prefix) - { + while !server.received_routes().iter().any(|e| e.prefix == prefix) { tokio::time::sleep(Duration::from_millis(20)).await; } }) @@ -393,7 +386,7 @@ async fn test_server_accepts_ipv6_update() -> Result<()> { #[tokio::test] async fn test_rib_records_ipv4_announce() -> Result<()> { let server = TestServer::start(65000).await?; - let mut events = server.adj_rib_in.subscribe(); + let mut events = server.route_changes(); let mut peer = TestPeer::connect(server.addr).await?; peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) @@ -409,11 +402,11 @@ async fn test_rib_records_ipv4_announce() -> Result<()> { let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??; assert!( - matches!(&event, RibEvent::Announced { prefix: p, .. } if p == &prefix), + matches!(&event, RouteChange::Announced { prefix: p, .. } if p == &prefix), "Expected Announced for 10.0.0.0/8, got {event:?}", ); - let snapshot = server.adj_rib_in.snapshot().await.unwrap(); + let snapshot = server.received_routes(); assert!( snapshot.iter().any(|e| e.prefix == prefix), "10.0.0.0/8 missing from RIB snapshot", @@ -425,7 +418,7 @@ async fn test_rib_records_ipv4_announce() -> Result<()> { #[tokio::test] async fn test_rib_records_ipv4_withdraw() -> Result<()> { let server = TestServer::start(65000).await?; - let mut events = server.adj_rib_in.subscribe(); + let mut events = server.route_changes(); let mut peer = TestPeer::connect(server.addr).await?; peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) @@ -452,11 +445,11 @@ async fn test_rib_records_ipv4_withdraw() -> Result<()> { let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??; assert!( - matches!(&event, RibEvent::Withdrawn { prefix: p, .. } if p == &prefix), + matches!(&event, RouteChange::Withdrawn { prefix: p, .. } if p == &prefix), "Expected Withdrawn for 10.0.0.0/8, got {event:?}", ); - let snapshot = server.adj_rib_in.snapshot().await.unwrap(); + let snapshot = server.received_routes(); assert!( !snapshot.iter().any(|e| e.prefix == prefix), "10.0.0.0/8 should have been removed from RIB snapshot", @@ -468,7 +461,7 @@ async fn test_rib_records_ipv4_withdraw() -> Result<()> { #[tokio::test] async fn test_rib_records_ipv6_announce() -> Result<()> { let server = TestServer::start(65000).await?; - let mut events = server.adj_rib_in.subscribe(); + let mut events = server.route_changes(); let mut peer = TestPeer::connect(server.addr).await?; peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) @@ -491,11 +484,11 @@ async fn test_rib_records_ipv6_announce() -> Result<()> { let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??; assert!( - matches!(&event, RibEvent::Announced { prefix: p, .. } if p == &prefix), + matches!(&event, RouteChange::Announced { prefix: p, .. } if p == &prefix), "Expected Announced for 2001:db8::/32, got {event:?}", ); - let snapshot = server.adj_rib_in.snapshot().await.unwrap(); + let snapshot = server.received_routes(); assert!( snapshot.iter().any(|e| e.prefix == prefix), "2001:db8::/32 missing from RIB snapshot", @@ -507,7 +500,7 @@ async fn test_rib_records_ipv6_announce() -> Result<()> { #[tokio::test] async fn test_rib_records_ipv6_withdraw() -> Result<()> { let server = TestServer::start(65000).await?; - let mut events = server.adj_rib_in.subscribe(); + let mut events = server.route_changes(); let mut peer = TestPeer::connect(server.addr).await?; peer.handshake(65001, Ipv4Addr::new(10, 0, 0, 1), 90) @@ -545,11 +538,11 @@ async fn test_rib_records_ipv6_withdraw() -> Result<()> { let event = tokio::time::timeout(Duration::from_secs(1), events.recv()).await??; assert!( - matches!(&event, RibEvent::Withdrawn { prefix: p, .. } if p == &prefix), + matches!(&event, RouteChange::Withdrawn { prefix: p, .. } if p == &prefix), "Expected Withdrawn for 2001:db8::/32, got {event:?}", ); - let snapshot = server.adj_rib_in.snapshot().await.unwrap(); + let snapshot = server.received_routes(); assert!( !snapshot.iter().any(|e| e.prefix == prefix), "2001:db8::/32 should have been removed from RIB snapshot", diff --git a/crates/integration-tests/tests/best_route.rs b/crates/integration-tests/tests/best_route.rs new file mode 100644 index 0000000..36db847 --- /dev/null +++ b/crates/integration-tests/tests/best_route.rs @@ -0,0 +1,239 @@ +//! End-to-end tests of route selection: routes for one prefix from several +//! peers, which of them the server chooses, and how that is shown. + +use std::net::Ipv4Addr; +use std::time::Duration; + +use bgp_integration_tests::test_peer::TestPeer; +use bgp_integration_tests::test_server::{RouteChange, TestServer}; +use bgp_packet::message::{ + AsPathAttribute, Message, NextHopPathAttribute, OriginPathAttribute, PathAttribute, + UpdateMessage, +}; +use bgp_primitives::ip_prefix::IpPrefix; +use eyre::Result; +use serde_json::Value; + +const LOCAL_ASN: u32 = 65000; +const TIMEOUT: Duration = Duration::from_secs(5); + +fn prefix(s: &str) -> IpPrefix { + IpPrefix::try_from(s).unwrap() +} + +/// A session from the peer numbered `number`, in AS 65000 + `number`. +async fn peer(server: &TestServer, number: u8) -> Result { + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake( + LOCAL_ASN + u32::from(number), + Ipv4Addr::new(10, 0, 0, number), + 90, + ) + .await?; + Ok(peer) +} + +fn announcement(prefixes: &[IpPrefix], as_path: &[u32]) -> Message { + Message::Update(UpdateMessage { + withdrawn_routes: vec![], + path_attributes: vec![ + PathAttribute::Origin(OriginPathAttribute::IGP), + AsPathAttribute::from_asns(as_path.to_vec()), + PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))), + ], + announced_routes: prefixes.to_vec(), + }) +} + +fn withdrawal(prefixes: &[IpPrefix]) -> Message { + Message::Update(UpdateMessage { + withdrawn_routes: prefixes.to_vec(), + path_attributes: vec![], + announced_routes: vec![], + }) +} + +/// Waits until the table holds `routes` routes. +async fn wait_for_routes(server: &TestServer, routes: usize) -> Result<()> { + tokio::time::timeout(TIMEOUT, async { + while server.rib.len() != routes { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await?; + Ok(()) +} + +/// The AS path of the route the server holds as best for `prefix`. +fn best_path(server: &TestServer, prefix: &IpPrefix) -> Option> { + let best = server + .received_routes() + .into_iter() + .find(|route| route.prefix == *prefix && route.best)?; + best.attributes + .iter() + .find_map(|attribute| match attribute { + PathAttribute::AsPath(path) => Some(path.segments[0].path.clone()), + _ => None, + }) +} + +async fn get_json(client: &reqwest::Client, url: &str) -> Result { + Ok(client + .get(url) + .send() + .await? + .error_for_status()? + .json() + .await?) +} + +/// Two peers offer the same prefix. The route with the shorter AS path is +/// chosen; when it goes, the other is. +#[tokio::test] +async fn test_the_best_route_is_chosen_and_replaced_when_it_goes() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let (base, _http) = server.serve_http().await?; + let client = reqwest::Client::new(); + let mut changes = server.route_changes(); + let p = prefix("203.0.113.0/24"); + + let mut far = peer(&server, 1).await?; + far.send(&announcement(&[p], &[65001, 64999, 64998])) + .await?; + wait_for_routes(&server, 1).await?; + assert_eq!(best_path(&server, &p), Some(vec![65001, 64999, 64998])); + + let mut near = peer(&server, 2).await?; + near.send(&announcement(&[p], &[65002, 64998])).await?; + wait_for_routes(&server, 2).await?; + assert_eq!(best_path(&server, &p), Some(vec![65002, 64998])); + + // Both routes are kept and shown; one is marked. + let status = get_json(&client, &format!("{base}/api/status")).await?; + assert_eq!(status["received_routes"], 2); + assert_eq!(status["received_prefixes"], 1); + let routes = get_json(&client, &format!("{base}/api/routes/received")).await?; + let best: Vec<&Value> = routes + .as_array() + .unwrap() + .iter() + .filter(|route| route["best"] == true) + .collect(); + assert_eq!(routes.as_array().unwrap().len(), 2); + assert_eq!(best.len(), 1); + assert_eq!(best[0]["peer_asn"], 65002); + + // The session with the best route ends: the other route takes over. + drop(near); + wait_for_routes(&server, 1).await?; + assert_eq!(best_path(&server, &p), Some(vec![65001, 64999, 64998])); + + far.send(&withdrawal(&[p])).await?; + wait_for_routes(&server, 0).await?; + assert_eq!(best_path(&server, &p), None); + + // A follower of the best routes is told of each change: the first + // route, the better one, back to the first, none. + let mut seen = Vec::new(); + while seen.last() != Some(&RouteChange::Withdrawn { prefix: p }) { + seen.push(tokio::time::timeout(TIMEOUT, changes.recv()).await??); + } + assert!( + seen.iter() + .all(|change| matches!(change, RouteChange::Announced { prefix, .. } | RouteChange::Withdrawn { prefix } if *prefix == p)), + "{seen:?}" + ); + Ok(()) +} + +/// The event stream reports changes of best route, and only those. +#[tokio::test] +async fn test_changes_of_best_route_are_on_the_event_stream() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let (base, _http) = server.serve_http().await?; + let p = prefix("203.0.113.0/24"); + + // A route that is there before the client connects is not an event. + let mut first = peer(&server, 1).await?; + first + .send(&announcement(&[prefix("198.51.100.0/24")], &[65001])) + .await?; + wait_for_routes(&server, 1).await?; + + let mut stream = reqwest::Client::new() + .get(format!("{base}/api/events")) + .send() + .await? + .error_for_status()?; + + first.send(&announcement(&[p], &[65001, 64998])).await?; + wait_for_routes(&server, 2).await?; + first.send(&withdrawal(&[p])).await?; + wait_for_routes(&server, 1).await?; + + let mut text = String::new(); + let mut events: Vec = Vec::new(); + tokio::time::timeout(TIMEOUT, async { + while !events.iter().any(|event| event["type"] == "best_withdrawn") { + let chunk = stream + .chunk() + .await? + .ok_or_else(|| eyre::eyre!("stream ended"))?; + text.push_str(std::str::from_utf8(&chunk)?); + events = text + .lines() + .filter_map(|line| line.strip_prefix("data:")) + .filter_map(|data| serde_json::from_str(data.trim()).ok()) + .collect(); + } + Ok::<_, eyre::Error>(()) + }) + .await??; + + let about_routes: Vec<&Value> = events + .iter() + .filter(|event| { + event["type"] + .as_str() + .is_some_and(|t| t.starts_with("best_")) + }) + .collect(); + // The announcement and the withdrawal may have been read as one change + // or as two; either way nothing is said of the route that was there + // from the start, and the last word is that the prefix is gone. + assert!(about_routes.len() <= 2, "{about_routes:?}"); + for event in &about_routes { + let named = event["prefix"] + .as_str() + .or(event["route"]["prefix"].as_str()); + assert_eq!(named, Some("203.0.113.0/24"), "{event}"); + } + assert_eq!(about_routes.last().unwrap()["type"], "best_withdrawn"); + Ok(()) +} + +/// The table can be far too large to send whole: the API gives a part of +/// it, and says how much there is. +#[tokio::test] +async fn test_received_routes_are_returned_up_to_a_limit() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let (base, _http) = server.serve_http().await?; + let client = reqwest::Client::new(); + + let prefixes: Vec = (0..40).map(|n| prefix(&format!("10.{n}.0.0/16"))).collect(); + let mut peer = peer(&server, 1).await?; + peer.send(&announcement(&prefixes, &[65001])).await?; + wait_for_routes(&server, 40).await?; + + let count = |routes: Value| routes.as_array().unwrap().len(); + let url = format!("{base}/api/routes/received"); + assert_eq!(count(get_json(&client, &url).await?), 40); + assert_eq!( + count(get_json(&client, &format!("{url}?limit=7")).await?), + 7 + ); + let status = get_json(&client, &format!("{base}/api/status")).await?; + assert_eq!(status["received_routes"], 40); + Ok(()) +} diff --git a/crates/integration-tests/tests/collision.rs b/crates/integration-tests/tests/collision.rs index aeffc9c..bd030ae 100644 --- a/crates/integration-tests/tests/collision.rs +++ b/crates/integration-tests/tests/collision.rs @@ -240,7 +240,7 @@ async fn test_restarted_peer_replaces_stale_session_when_enabled() -> Result<()> })) .await?; tokio::time::timeout(TIMEOUT, async { - while server.adj_rib_in.snapshot().await.unwrap().is_empty() { + while server.received_routes().is_empty() { tokio::time::sleep(Duration::from_millis(10)).await; } }) @@ -265,7 +265,7 @@ async fn test_restarted_peer_replaces_stale_session_when_enabled() -> Result<()> }) .await?; assert!( - server.adj_rib_in.snapshot().await.unwrap().is_empty(), + server.received_routes().is_empty(), "stale routes were not flushed" ); Ok(()) diff --git a/crates/integration-tests/tests/failure.rs b/crates/integration-tests/tests/failure.rs index 0a49ebd..47fd29f 100644 --- a/crates/integration-tests/tests/failure.rs +++ b/crates/integration-tests/tests/failure.rs @@ -1,9 +1,9 @@ //! What the server does when a part of it that it cannot do without fails. //! -//! A RIB's actor stops only by panicking. Before, nothing noticed: sessions -//! stayed up, routes sent by peers were dropped on the floor, and the status -//! page showed an empty table. Now the server tells its peers it is going -//! and stops with an error, for whatever supervises it to restart it. +//! The local RIB's actor stops only by panicking. Before, nothing noticed: +//! sessions stayed up, nothing was announced any more, and the status page +//! showed an empty table. Now the server tells its peers it is going and +//! stops with an error, for whatever supervises it to restart it. use std::net::Ipv4Addr; use std::time::Duration; @@ -47,49 +47,10 @@ async fn farewell(peer: &mut TestPeer) -> Result { .await? } -#[tokio::test] -async fn test_losing_the_adj_rib_in_stops_the_server_and_tells_the_peers() -> Result<()> { - let mut server = TestServer::start(LOCAL_ASN).await?; - let (base, _http) = server.serve_http().await?; - let mut peer = established(&server).await?; - - server.adj_rib_in.stop_for_testing(); - - // Nothing has used the table since; the server notices by itself. - let outcome = tokio::time::timeout(TIMEOUT, server.stopped()).await?; - assert!( - matches!(outcome, Err(ServerError::RibStopped(RibStopped { rib })) if rib == "Adj-RIB-In"), - "{outcome:?}" - ); - assert_eq!( - farewell(&mut peer).await?, - BgpError::Cease(CeaseSubcode::AdminShutdown) - ); - - // The status interface says the table cannot be read, where it used to - // show an empty one. - let client = reqwest::Client::new(); - for path in ["/api/status", "/api/routes/received", "/metricz"] { - let response = client.get(format!("{base}{path}")).send().await?; - assert_eq!(response.status(), 503, "{path}"); - assert_eq!( - response.text().await?, - "the Adj-RIB-In has stopped", - "{path}" - ); - } - // What does not need the table still answers. - let response = client - .get(format!("{base}/api/routes/local")) - .send() - .await?; - assert_eq!(response.status(), 200); - Ok(()) -} - #[tokio::test] async fn test_losing_the_local_rib_stops_the_server_and_tells_the_peers() -> Result<()> { let mut server = TestServer::start(LOCAL_ASN).await?; + let (base, _http) = server.serve_http().await?; let mut peer = established(&server).await?; server.local_rib.stop_for_testing(); @@ -103,5 +64,24 @@ async fn test_losing_the_local_rib_stops_the_server_and_tells_the_peers() -> Res farewell(&mut peer).await?, BgpError::Cease(CeaseSubcode::AdminShutdown) ); + + // The status interface says the table cannot be read, where it used to + // show an empty one. + let client = reqwest::Client::new(); + for path in ["/api/status", "/api/routes/local", "/metricz"] { + let response = client.get(format!("{base}{path}")).send().await?; + assert_eq!(response.status(), 503, "{path}"); + assert_eq!( + response.text().await?, + "the local RIB has stopped", + "{path}" + ); + } + // What does not need that table still answers. + let response = client + .get(format!("{base}/api/routes/received")) + .send() + .await?; + assert_eq!(response.status(), 200); Ok(()) } diff --git a/crates/integration-tests/tests/http_api.rs b/crates/integration-tests/tests/http_api.rs index e602817..00e1493 100644 --- a/crates/integration-tests/tests/http_api.rs +++ b/crates/integration-tests/tests/http_api.rs @@ -298,6 +298,6 @@ async fn test_metricz() -> Result<()> { "{text}" ); assert!(metric("bgp_local_routes").is_some(), "{text}"); - assert!(metric("bgp_adj_rib_in_routes").is_some(), "{text}"); + assert!(metric("bgp_rib_routes").is_some(), "{text}"); Ok(()) } diff --git a/crates/integration-tests/tests/interop.rs b/crates/integration-tests/tests/interop.rs index fd10dd3..12ef78f 100644 --- a/crates/integration-tests/tests/interop.rs +++ b/crates/integration-tests/tests/interop.rs @@ -105,7 +105,7 @@ async fn test_bird_open_and_end_of_rib_are_accepted() -> Result<()> { 1, "session dropped after End-of-RIB" ); - assert!(server.adj_rib_in.snapshot().await.unwrap().is_empty()); + assert!(server.received_routes().is_empty()); Ok(()) } diff --git a/crates/integration-tests/tests/negotiation.rs b/crates/integration-tests/tests/negotiation.rs index 5c76703..e1c2ecf 100644 --- a/crates/integration-tests/tests/negotiation.rs +++ b/crates/integration-tests/tests/negotiation.rs @@ -120,7 +120,7 @@ async fn test_routes_of_a_family_not_negotiated_are_not_accepted() -> Result<()> let routes = tokio::time::timeout(TIMEOUT, async { loop { - let routes = server.adj_rib_in.snapshot().await.unwrap(); + let routes = server.received_routes(); if !routes.is_empty() { return routes; } @@ -304,7 +304,7 @@ async fn test_a_peer_without_capabilities_gets_two_octet_as_paths() -> Result<() .await?; let routes = tokio::time::timeout(TIMEOUT, async { loop { - let routes = server.adj_rib_in.snapshot().await.unwrap(); + let routes = server.received_routes(); if !routes.is_empty() { return routes; } @@ -314,8 +314,7 @@ async fn test_a_peer_without_capabilities_gets_two_octet_as_paths() -> Result<() .await?; assert!( routes[0] - .entry - .path_attributes + .attributes .contains(&AsPathAttribute::from_asns(vec![PEER_ASN, 64512])) ); Ok(()) diff --git a/crates/integration-tests/tests/update_errors.rs b/crates/integration-tests/tests/update_errors.rs index f48a93d..46dfe99 100644 --- a/crates/integration-tests/tests/update_errors.rs +++ b/crates/integration-tests/tests/update_errors.rs @@ -5,11 +5,10 @@ use std::net::Ipv4Addr; use std::time::Duration; use bgp_integration_tests::test_peer::TestPeer; -use bgp_integration_tests::test_server::TestServer; +use bgp_integration_tests::test_server::{ReceivedRoute, TestServer}; use bgp_packet::errors::{BgpError, UpdateMsgSubcode}; use bgp_packet::message::{LocalPrefPathAttribute, Message, PathAttribute}; use bgp_primitives::ip_prefix::IpPrefix; -use bgp_rib::SnapshotEntry; use bgp_server::metrics::UPDATE_ERRORS; use eyre::{Result, bail}; @@ -70,12 +69,12 @@ async fn established(server: &TestServer, asn: u32) -> Result { } /// Waits until the server's Adj-RIB-In holds exactly `prefixes`. -async fn wait_for_routes(server: &TestServer, prefixes: &[&str]) -> Result> { +async fn wait_for_routes(server: &TestServer, prefixes: &[&str]) -> Result> { let mut expected: Vec = prefixes.iter().map(|p| prefix(p)).collect(); expected.sort_by_key(|p| p.to_string()); Ok(tokio::time::timeout(TIMEOUT, async { loop { - let mut routes = server.adj_rib_in.snapshot().await.unwrap(); + let mut routes = server.received_routes(); routes.sort_by_key(|r| r.prefix.to_string()); if routes.iter().map(|r| &r.prefix).eq(expected.iter()) { return routes; @@ -154,11 +153,10 @@ async fn test_a_malformed_aggregator_is_dropped_and_the_route_kept() -> Result<( peer.send_raw(&update(&attributes, &ROUTE_A)).await?; let routes = wait_for_routes(&server, &["10.1.0.0/16"]).await?; - assert_eq!(routes[0].entry.path_attributes.len(), 3); + assert_eq!(routes[0].attributes.len(), 3); assert!( !routes[0] - .entry - .path_attributes + .attributes .iter() .any(|a| matches!(a, PathAttribute::Aggregator(_))) ); @@ -170,10 +168,9 @@ async fn test_a_malformed_aggregator_is_dropped_and_the_route_kept() -> Result<( #[tokio::test] async fn test_local_pref_is_kept_only_from_an_internal_peer() -> Result<()> { let local_pref = attr(WELL_KNOWN, 5, &[0, 0, 0, 200]); - let has_local_pref = |routes: &[SnapshotEntry]| { + let has_local_pref = |routes: &[ReceivedRoute]| { routes[0] - .entry - .path_attributes + .attributes .contains(&PathAttribute::LocalPref(LocalPrefPathAttribute(200))) }; diff --git a/crates/packet/src/message.rs b/crates/packet/src/message.rs index 84b95c8..281eb2e 100644 --- a/crates/packet/src/message.rs +++ b/crates/packet/src/message.rs @@ -67,17 +67,15 @@ impl Message { /// handles without resetting the session have been handled. Use /// [`Message::decode_with_report`] to learn of them. pub fn decode(ctx: &ParserContext, buf: &[u8]) -> Result { - Self::decode_with_report(ctx, buf).map(|(msg, _)| msg) + Self::decode_with_report(ctx, buf).map(|decoded| decoded.message) } /// As [`Message::decode`], and also returns the errors in an UPDATE /// that were handled without resetting the session. The list is empty /// for a well-formed UPDATE and for every other message. - pub fn decode_with_report( - ctx: &ParserContext, - buf: &[u8], - ) -> Result<(Self, Vec), DecodeError> { + pub fn decode_with_report(ctx: &ParserContext, buf: &[u8]) -> Result { let mut update_errors = Vec::new(); + let mut attributes_wire = None; let header_error = |subcode, data: Vec, detail: String| DecodeError { notification: BgpError::MsgHeader(subcode), data, @@ -154,6 +152,7 @@ impl Message { 2 => { let decoded = UpdateMessage::decode(ctx, body)?; update_errors = decoded.errors; + attributes_wire = decoded.attributes_wire; Ok(Self::Update(decoded.update)) } 3 => NotificationMessage::from_wire(ctx, body) @@ -175,7 +174,11 @@ impl Message { format!("unknown BGP message type {other}"), )), }?; - Ok((message, update_errors)) + Ok(Decoded { + message, + update_errors, + attributes_wire, + }) } /// Parses a complete BGP message including the 19-byte header from `buf`. @@ -249,6 +252,18 @@ impl Message { } } +/// A message, and what else decoding it found out. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Decoded { + pub message: Message, + /// The errors in an UPDATE that were handled without resetting the + /// session (RFC 7606). Empty for a well-formed UPDATE and for every + /// other message. + pub update_errors: Vec, + /// See [`crate::update_decode::DecodedUpdate::attributes_wire`]. + pub attributes_wire: Option>, +} + /// A request to send again every route of one address family (RFC 2918). #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub struct RouteRefreshMessage { diff --git a/crates/packet/src/update_decode.rs b/crates/packet/src/update_decode.rs index f0112a7..28eeb5d 100644 --- a/crates/packet/src/update_decode.rs +++ b/crates/packet/src/update_decode.rs @@ -65,6 +65,15 @@ pub struct DecodedUpdate { pub update: UpdateMessage, /// Empty when the UPDATE was well formed. pub errors: Vec, + /// The attributes in `update`, as they were on the wire, for a table + /// that keeps attributes in that form and need not write them out + /// again. MP_REACH_NLRI and MP_UNREACH_NLRI are left out, since they + /// name prefixes, and so is every attribute that was discarded. + /// + /// `None` when there are no routes for them to belong to, or when the + /// bytes are not in the one form such a table keeps: on a session with + /// two-octet AS numbers. + pub attributes_wire: Option>, } const ORIGIN: u8 = 1; @@ -160,7 +169,8 @@ impl UpdateMessage { let announced_routes = prefixes("NLRI", nlri_buf)?; let mut errors = Vec::new(); - let path_attributes = attributes(ctx, attributes_buf, &mut errors)?; + let mut kept_wire = Vec::with_capacity(attributes_buf.len()); + let path_attributes = attributes(ctx, attributes_buf, &mut errors, &mut kept_wire)?; let mut update = UpdateMessage { withdrawn_routes, path_attributes, @@ -178,21 +188,31 @@ impl UpdateMessage { detail: format!("mandatory attribute {missing} is missing"), }); } - if errors + let withdrawn = errors .iter() - .any(|e| e.handling == Handling::TreatAsWithdraw) - { + .any(|e| e.handling == Handling::TreatAsWithdraw); + if withdrawn { update = as_withdrawal(update); } - Ok(DecodedUpdate { update, errors }) + let attributes_wire = + (!withdrawn && ctx.four_octet_asn == Some(true) && !kept_wire.is_empty()) + .then_some(kept_wire); + Ok(DecodedUpdate { + update, + errors, + attributes_wire, + }) } } -/// Reads the path attributes, leaving out the ones that are discarded. +/// Reads the path attributes, leaving out the ones that are discarded. The +/// bytes of the ones that are kept and are a property of a route (all but +/// MP_REACH_NLRI and MP_UNREACH_NLRI) are added to `kept_wire`. fn attributes( ctx: &ParserContext, mut buf: &[u8], errors: &mut Vec, + kept_wire: &mut Vec, ) -> Result, DecodeError> { let mut attributes = Vec::new(); let mut seen = HashSet::new(); @@ -283,6 +303,7 @@ fn attributes( type_code, payload: value.to_vec(), }); + kept_wire.extend_from_slice(attribute_wire); continue; }; let flags_are_wrong = flags.optional() != optional || flags.transitive() != transitive; @@ -349,7 +370,13 @@ fn attributes( } }; match parsed { - Ok(attribute) => attributes.push(attribute), + Ok(attribute) => { + // With wrong flags it was read only to find its routes. + if !flags_are_wrong && !matches!(type_code, MP_REACH_NLRI | MP_UNREACH_NLRI) { + kept_wire.extend_from_slice(attribute_wire); + } + attributes.push(attribute); + } Err(detail) => match handling_of_malformed(type_code) { Some(handling) => error( handling, @@ -562,6 +589,43 @@ mod tests { ); } + #[test] + fn the_kept_attributes_are_handed_over_as_they_were_on_the_wire() { + // All of a well-formed UPDATE's attributes, but not MP_REACH_NLRI. + let attributes = vec![mp_reach_v6(), origin(), as_path(), next_hop()]; + let decoded = UpdateMessage::decode(&ebgp(), &body(&[], &attributes, &NLRI)).unwrap(); + assert_eq!( + decoded.attributes_wire, + Some([origin(), as_path(), next_hop()].concat()) + ); + + // Not an attribute that was discarded. + let decoded = announce_with(&ebgp(), local_pref()); + assert_eq!(decoded.attributes_wire, Some(mandatory().concat())); + + // Nothing when the routes were withdrawn instead: there is no route + // for the attributes to belong to. + let decoded = announce_with(&ebgp(), attr(OPTIONAL, MED, &[1])); + assert_eq!(decoded.attributes_wire, None); + let decoded = UpdateMessage::decode(&ebgp(), &body(&NLRI, &[], &[])).unwrap(); + assert_eq!(decoded.attributes_wire, None); + + // Nor on a session with two-octet AS numbers, where the bytes are + // not in the form a table keeps. + let two_octet = ParserContext { + four_octet_asn: Some(false), + ibgp: false, + }; + let attributes = vec![ + origin(), + attr(WELL_KNOWN, AS_PATH, &[2, 1, 0xfd, 0xe9]), + next_hop(), + ]; + let decoded = UpdateMessage::decode(&two_octet, &body(&[], &attributes, &NLRI)).unwrap(); + assert!(decoded.errors.is_empty()); + assert_eq!(decoded.attributes_wire, None); + } + #[test] fn an_update_that_announces_nothing_needs_no_attributes() { // End-of-RIB, and a plain withdrawal. diff --git a/crates/packet/tests/properties.rs b/crates/packet/tests/properties.rs index 30b753c..4316354 100644 --- a/crates/packet/tests/properties.rs +++ b/crates/packet/tests/properties.rs @@ -454,9 +454,10 @@ fn session_and_message() -> impl Strategy { /// accepts: it can be shown and written, and unless the decoder had to /// repair it, it reads back as itself. fn check_decoding(ctx: &ParserContext, wire: &[u8]) -> Result<(), TestCaseError> { - let Ok((msg, repairs)) = Message::decode_with_report(ctx, wire) else { + let Ok(decoded) = Message::decode_with_report(ctx, wire) else { return Ok(()); }; + let (msg, repairs) = (decoded.message, decoded.update_errors); // Display of what a peer sent must not be able to panic: it is logged. let _ = format!("{msg:?} {repairs:?}"); if let Message::Update(update) = &msg { @@ -494,10 +495,10 @@ fn check_decoding(ctx: &ParserContext, wire: &[u8]) -> Result<(), TestCaseError> } prop_assert_eq!(usize::from(msg.wire_len(ctx).unwrap()), rewritten.len()); if repairs.is_empty() { - let (again, repairs) = Message::decode_with_report(ctx, &rewritten) + let again = Message::decode_with_report(ctx, &rewritten) .map_err(|e| TestCaseError::fail(format!("written but cannot be read: {e}")))?; - prop_assert!(repairs.is_empty(), "{:?}", repairs); - prop_assert_eq!(again, msg); + prop_assert!(again.update_errors.is_empty(), "{:?}", again.update_errors); + prop_assert_eq!(again.message, msg); } Ok(()) } @@ -595,11 +596,11 @@ proptest! { }; prop_assert_eq!(usize::from(msg.wire_len(&ctx).unwrap()), wire.len()); - let (decoded, repairs) = Message::decode_with_report(&ctx, &wire) + let decoded = Message::decode_with_report(&ctx, &wire) .map_err(|e| TestCaseError::fail(format!("{e}")))?; // What we write is well formed: nothing for the decoder to repair. - prop_assert!(repairs.is_empty(), "{:?}", repairs); - prop_assert_eq!(&decoded, &msg); + prop_assert!(decoded.update_errors.is_empty(), "{:?}", decoded.update_errors); + prop_assert_eq!(&decoded.message, &msg); let (rest, _) = Message::from_wire(&ctx, &wire).unwrap(); prop_assert!(rest.is_empty()); } diff --git a/crates/rib/src/adj_rib_in.rs b/crates/rib/src/adj_rib_in.rs deleted file mode 100644 index 9c29a92..0000000 --- a/crates/rib/src/adj_rib_in.rs +++ /dev/null @@ -1,277 +0,0 @@ -//! The Adj-RIB-In: routes as peers announced them, before any policy. -//! -//! One actor owns the table. Handles are cheap to clone and talk to it over -//! a channel; changes are published as [`RibEvent`]s. - -use std::collections::HashMap; - -use bgp_primitives::ip_prefix::IpPrefix; -use tokio::sync::{broadcast, mpsc, oneshot}; -use tokio::task::AbortHandle; - -use crate::entry::{PeerId, ReceivedUpdate, RibEntry, SnapshotEntry}; -use crate::error::RibStopped; -use crate::event::RibEvent; - -const EVENT_CHANNEL_CAPACITY: usize = 1024; -const CMD_CHANNEL_CAPACITY: usize = 256; - -enum RibCommand { - /// Apply one UPDATE from `peer`. - Apply { - peer: PeerId, - update: ReceivedUpdate, - }, - /// Drop everything learned from `peer`, e.g. because its session ended. - RemovePeer { - peer: PeerId, - reply: oneshot::Sender, - }, - Snapshot { - reply: oneshot::Sender>, - }, - Len { - reply: oneshot::Sender, - }, -} - -const STOPPED: RibStopped = RibStopped { rib: "Adj-RIB-In" }; - -/// A cheaply-cloneable handle to the Adj-RIB-In actor. -/// -/// Every operation fails with [`RibStopped`] if the actor is gone; see -/// there for what that means. -#[derive(Clone)] -pub struct AdjRibIn { - tx: mpsc::Sender, - events: broadcast::Sender, - actor: AbortHandle, -} - -impl AdjRibIn { - /// Spawns the backing actor and returns a handle. - pub fn spawn() -> Self { - let (tx, rx) = mpsc::channel(CMD_CHANNEL_CAPACITY); - let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); - let actor = tokio::spawn(run_actor(rx, events.clone())).abort_handle(); - Self { tx, events, actor } - } - - /// Applies one UPDATE received from `peer`: its withdrawals first, then - /// its announcements, so a prefix named in both ends up announced - /// (RFC 4271 §4.3). - /// - /// Returns once the update is queued for the actor, which applies - /// updates in the order they were queued. - pub async fn apply(&self, peer: PeerId, update: ReceivedUpdate) -> Result<(), RibStopped> { - self.tx - .send(RibCommand::Apply { peer, update }) - .await - .map_err(|_| STOPPED) - } - - /// Removes every route learned from `peer`, emitting a - /// [`RibEvent::Withdrawn`] for each (RFC 4271 §8.2.2: routes from a - /// terminated session are deleted). Returns how many were removed. - pub async fn remove_peer(&self, peer: PeerId) -> Result { - self.ask(|reply| RibCommand::RemovePeer { peer, reply }) - .await - } - - /// Returns a point-in-time snapshot of all routes in the RIB. - pub async fn snapshot(&self) -> Result, RibStopped> { - self.ask(|reply| RibCommand::Snapshot { reply }).await - } - - /// How many routes the RIB holds, without copying them. - pub async fn len(&self) -> Result { - self.ask(|reply| RibCommand::Len { reply }).await - } - - /// Whether the RIB holds no routes. - pub async fn is_empty(&self) -> Result { - Ok(self.len().await? == 0) - } - - /// Sends a command that is answered, and waits for the answer. - async fn ask( - &self, - command: impl FnOnce(oneshot::Sender) -> RibCommand, - ) -> Result { - let (reply, answer) = oneshot::channel(); - self.tx.send(command(reply)).await.map_err(|_| STOPPED)?; - answer.await.map_err(|_| STOPPED) - } - - /// Resolves when the actor has stopped. It never does in a healthy - /// process; whoever owns the RIB waits on this to notice a failure even - /// when nothing happens to be using the table. - pub async fn stopped(&self) { - self.tx.closed().await - } - - /// Stops the actor, as a panic in it would. For tests of how the rest - /// of the server copes. - #[doc(hidden)] - pub fn stop_for_testing(&self) { - self.actor.abort(); - } - - /// Subscribe to a stream of RIB mutation events. - /// - /// The receiver will see every [`RibEvent`] from the point of subscription. - /// If the consumer falls behind by more than [`EVENT_CHANNEL_CAPACITY`] - /// events it will receive a [`broadcast::error::RecvError::Lagged`] error. - pub fn subscribe(&self) -> broadcast::Receiver { - self.events.subscribe() - } -} - -async fn run_actor(mut rx: mpsc::Receiver, events: broadcast::Sender) { - // Keyed by peer first, so that dropping a peer's routes when its session - // ends touches only that peer's table. - let mut table: HashMap> = HashMap::new(); - - while let Some(cmd) = rx.recv().await { - match cmd { - // Event send errors below mean there are no subscribers, which is fine. - RibCommand::Apply { peer, update } => { - let routes = table.entry(peer).or_default(); - for prefix in update.withdrawn { - if routes.remove(&prefix).is_some() { - let _ = events.send(RibEvent::Withdrawn { peer, prefix }); - } - } - for (prefix, entry) in update.announced { - routes.insert(prefix, entry.clone()); - let _ = events.send(RibEvent::Announced { - peer, - prefix, - entry, - }); - } - if routes.is_empty() { - table.remove(&peer); - } - } - RibCommand::RemovePeer { peer, reply } => { - let routes = table.remove(&peer).unwrap_or_default(); - let removed = routes.len(); - for prefix in routes.into_keys() { - let _ = events.send(RibEvent::Withdrawn { peer, prefix }); - } - let _ = reply.send(removed); - } - RibCommand::Snapshot { reply } => { - let snap = table - .iter() - .flat_map(|(peer, routes)| { - routes.iter().map(|(prefix, entry)| SnapshotEntry { - peer: *peer, - prefix: *prefix, - entry: entry.clone(), - }) - }) - .collect(); - // A failed reply means the asker stopped waiting. - let _ = reply.send(snap); - } - RibCommand::Len { reply } => { - let _ = reply.send(table.values().map(HashMap::len).sum()); - } - } - } -} - -#[cfg(test)] -mod tests { - use std::sync::Arc; - use std::time::SystemTime; - - use super::*; - - fn prefix(s: &str) -> IpPrefix { - IpPrefix::try_from(s).unwrap() - } - - fn peer(port: u16, asn: u32) -> PeerId { - PeerId { - addr: ([192, 0, 2, 1], port).into(), - asn, - } - } - - fn announce(prefixes: &[&str]) -> ReceivedUpdate { - let entry = RibEntry { - path_attributes: Arc::from(vec![]), - next_hop: None, - received_at: SystemTime::now(), - }; - ReceivedUpdate { - withdrawn: vec![], - announced: prefixes - .iter() - .map(|p| (prefix(p), entry.clone())) - .collect(), - } - } - - #[tokio::test] - async fn a_prefix_withdrawn_and_announced_by_one_update_ends_up_announced() { - let rib = AdjRibIn::spawn(); - let a = peer(179, 65001); - rib.apply(a, announce(&["10.0.0.0/8"])).await.unwrap(); - - let mut update = announce(&["10.0.0.0/8"]); - update.withdrawn = vec![prefix("10.0.0.0/8")]; - rib.apply(a, update).await.unwrap(); - - assert_eq!(rib.snapshot().await.unwrap().len(), 1); - } - - #[tokio::test] - async fn removing_a_peer_leaves_other_peers_routes_alone() { - let rib = AdjRibIn::spawn(); - let (a, b) = (peer(179, 65001), peer(180, 65002)); - rib.apply(a, announce(&["10.0.0.0/8", "10.1.0.0/16"])) - .await - .unwrap(); - rib.apply(b, announce(&["10.0.0.0/8"])).await.unwrap(); - // Commands are handled in order, so once the snapshot is back the - // announcements above have been applied and their events sent. - assert_eq!(rib.snapshot().await.unwrap().len(), 3); - assert_eq!(rib.len().await, Ok(3)); - let mut events = rib.subscribe(); - - assert_eq!(rib.remove_peer(a).await, Ok(2)); - - let left = rib.snapshot().await.unwrap(); - assert_eq!(left.len(), 1); - assert_eq!(left[0].peer, b); - for _ in 0..2 { - assert!( - matches!(events.recv().await.unwrap(), RibEvent::Withdrawn { peer, .. } if peer == a) - ); - } - assert_eq!(rib.remove_peer(a).await, Ok(0), "nothing left to remove"); - } - - /// A stopped actor used to look like an empty table: updates vanished, - /// and a snapshot came back with no routes. - #[tokio::test] - async fn a_stopped_actor_is_an_error_and_not_an_empty_table() { - let rib = AdjRibIn::spawn(); - let a = peer(179, 65001); - rib.apply(a, announce(&["10.0.0.0/8"])).await.unwrap(); - assert_eq!(rib.len().await, Ok(1)); - - rib.stop_for_testing(); - rib.stopped().await; - - assert_eq!(rib.apply(a, announce(&["10.1.0.0/16"])).await, Err(STOPPED)); - assert_eq!(rib.snapshot().await.map(|s| s.len()), Err(STOPPED)); - assert_eq!(rib.len().await, Err(STOPPED)); - assert_eq!(rib.remove_peer(a).await, Err(STOPPED)); - assert_eq!(STOPPED.to_string(), "the Adj-RIB-In has stopped"); - } -} diff --git a/crates/rib/src/entry.rs b/crates/rib/src/entry.rs deleted file mode 100644 index 15c3e36..0000000 --- a/crates/rib/src/entry.rs +++ /dev/null @@ -1,197 +0,0 @@ -use std::net::SocketAddr; -use std::sync::Arc; -use std::time::SystemTime; - -use bgp_packet::message::{NlriNextHop, PathAttribute, UpdateMessage}; -use bgp_primitives::ip_prefix::IpPrefix; - -/// Identifies a BGP peer by its TCP address and negotiated ASN. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct PeerId { - pub addr: SocketAddr, - pub asn: u32, -} - -impl std::fmt::Display for PeerId { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}(AS{})", self.addr, self.asn) - } -} - -/// A single route stored in the Adj-RIB-In. -/// -/// Cloning is cheap: the attribute list is shared, not copied. -#[derive(Debug, Clone)] -pub struct RibEntry { - /// The attributes of the route. Every prefix announced by one UPDATE - /// shares one list. MP_REACH_NLRI and MP_UNREACH_NLRI are left out: they - /// carry prefixes, not properties of a route, and the one thing in them - /// that is a property, the next hop, is in `next_hop`. - pub path_attributes: Arc<[PathAttribute]>, - /// Next hop of a route that arrived in MP_REACH_NLRI. `None` for an IPv4 - /// route from the body of the UPDATE, whose next hop is the NEXT_HOP - /// attribute in `path_attributes`. - pub next_hop: Option, - pub received_at: SystemTime, -} - -/// A flattened row returned by [`crate::AdjRibIn::snapshot`]. -#[derive(Debug, Clone)] -pub struct SnapshotEntry { - pub peer: PeerId, - pub prefix: IpPrefix, - pub entry: RibEntry, -} - -/// What one UPDATE message changes in the Adj-RIB-In. -/// -/// An UPDATE names prefixes in four places: withdrawn routes and NLRI in the -/// message body (IPv4 only), and the MP_UNREACH_NLRI and MP_REACH_NLRI -/// attributes (every other family, RFC 4760). This gathers them into the -/// two lists the RIB acts on. -#[derive(Debug, Clone, Default)] -pub struct ReceivedUpdate { - pub withdrawn: Vec, - pub announced: Vec<(IpPrefix, RibEntry)>, -} - -impl ReceivedUpdate { - pub fn from_message(update: &UpdateMessage) -> Self { - Self::from_message_at(update, SystemTime::now()) - } - - fn from_message_at(update: &UpdateMessage, received_at: SystemTime) -> Self { - let mut withdrawn = update.withdrawn_routes.clone(); - // One allocation for the whole UPDATE, however many prefixes it has. - let shared: Arc<[PathAttribute]> = update - .path_attributes - .iter() - .filter(|attr| { - !matches!( - attr, - PathAttribute::MpReachNlri(_) | PathAttribute::MpUnreachNlri(_) - ) - }) - .cloned() - .collect(); - let entry = |next_hop| RibEntry { - path_attributes: Arc::clone(&shared), - next_hop, - received_at, - }; - - let mut announced: Vec<(IpPrefix, RibEntry)> = update - .announced_routes - .iter() - .map(|prefix| (*prefix, entry(None))) - .collect(); - for attr in &update.path_attributes { - match attr { - PathAttribute::MpReachNlri(mp) => announced.extend( - mp.prefixes - .iter() - .map(|prefix| (*prefix, entry(Some(mp.next_hop.clone())))), - ), - PathAttribute::MpUnreachNlri(mp) => withdrawn.extend(mp.prefixes.iter().cloned()), - _ => {} - } - } - Self { - withdrawn, - announced, - } - } -} - -#[cfg(test)] -mod tests { - use std::net::Ipv6Addr; - - use bgp_packet::message::{ - MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, OriginPathAttribute, - }; - use bgp_primitives::constants::{Afi, Safi}; - - use super::*; - - fn prefix(s: &str) -> IpPrefix { - IpPrefix::try_from(s).unwrap() - } - - fn mp_reach(prefixes: Vec) -> PathAttribute { - PathAttribute::MpReachNlri(MpReachNlriPathAttribute { - afi: Afi::IPV6, - safi: Safi::UNICAST, - next_hop: NlriNextHop::Ipv6(Ipv6Addr::LOCALHOST), - prefixes, - }) - } - - /// The Adj-RIB-In once kept a full copy of the attribute list per prefix, - /// MP_REACH_NLRI included, so an UPDATE with n prefixes stored n copies - /// of all n prefixes. - #[test] - fn prefixes_of_one_update_share_attributes_that_hold_no_prefixes() { - let prefixes: Vec = (0..500) - .map(|n| prefix(&format!("2001:db8:{n:x}::/48"))) - .collect(); - let update = UpdateMessage { - withdrawn_routes: vec![], - path_attributes: vec![ - PathAttribute::Origin(OriginPathAttribute::IGP), - mp_reach(prefixes.clone()), - ], - announced_routes: vec![], - }; - - let received = ReceivedUpdate::from_message(&update); - - assert_eq!(received.announced.len(), 500); - let first = &received.announced[0].1; - assert_eq!( - first.path_attributes[..], - [PathAttribute::Origin(OriginPathAttribute::IGP)] - ); - for (prefix, entry) in &received.announced { - assert!(prefixes.contains(prefix)); - assert!(Arc::ptr_eq(&entry.path_attributes, &first.path_attributes)); - assert_eq!(entry.next_hop, Some(NlriNextHop::Ipv6(Ipv6Addr::LOCALHOST))); - } - } - - #[test] - fn prefixes_are_gathered_from_the_body_and_the_multiprotocol_attributes() { - let update = UpdateMessage { - withdrawn_routes: vec![prefix("10.0.0.0/8")], - path_attributes: vec![ - PathAttribute::Origin(OriginPathAttribute::IGP), - mp_reach(vec![prefix("2001:db8:1::/48")]), - PathAttribute::MpUnreachNlri(MpUnreachNlriPathAttribute { - afi: Afi::IPV6, - safi: Safi::UNICAST, - prefixes: vec![prefix("2001:db8:2::/48")], - }), - ], - announced_routes: vec![prefix("192.0.2.0/24")], - }; - - let received = ReceivedUpdate::from_message(&update); - - assert_eq!( - received.withdrawn, - vec![prefix("10.0.0.0/8"), prefix("2001:db8:2::/48")] - ); - let announced: Vec<_> = received - .announced - .iter() - .map(|(p, e)| (*p, e.next_hop.is_some())) - .collect(); - assert_eq!( - announced, - vec![ - (prefix("192.0.2.0/24"), false), - (prefix("2001:db8:1::/48"), true) - ] - ); - } -} diff --git a/crates/rib/src/error.rs b/crates/rib/src/error.rs index 71982c6..9cc68f2 100644 --- a/crates/rib/src/error.rs +++ b/crates/rib/src/error.rs @@ -2,13 +2,13 @@ /// changed. /// /// The actor runs for as long as a handle to it exists; it stops early only -/// by panicking. That is a bug, and one the server cannot work around: -/// routes would be accepted from peers and go nowhere. Callers pass this up +/// by panicking. That is a bug, and one the server cannot work around: it +/// could no longer say what it originates. Callers pass this up /// until the process exits and is restarted, rather than carry on with a /// table that looks empty. #[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] #[error("the {rib} has stopped")] pub struct RibStopped { - /// Which table: "Adj-RIB-In" or "local RIB". + /// Which table. pub rib: &'static str, } diff --git a/crates/rib/src/event.rs b/crates/rib/src/event.rs deleted file mode 100644 index 3716b25..0000000 --- a/crates/rib/src/event.rs +++ /dev/null @@ -1,16 +0,0 @@ -use bgp_primitives::ip_prefix::IpPrefix; - -use crate::entry::{PeerId, RibEntry}; - -#[derive(Debug, Clone)] -pub enum RibEvent { - Announced { - peer: PeerId, - prefix: IpPrefix, - entry: RibEntry, - }, - Withdrawn { - peer: PeerId, - prefix: IpPrefix, - }, -} diff --git a/crates/rib/src/lib.rs b/crates/rib/src/lib.rs index a599934..61682b4 100644 --- a/crates/rib/src/lib.rs +++ b/crates/rib/src/lib.rs @@ -1,12 +1,11 @@ -mod adj_rib_in; -mod entry; +//! The router's tables: the routes it has learned and chosen +//! ([`table::RibTable`]) and the routes it originates ([`LocalRib`]). + mod error; -mod event; mod local_rib; +mod peer; pub mod table; -pub use adj_rib_in::AdjRibIn; -pub use entry::{PeerId, ReceivedUpdate, RibEntry, SnapshotEntry}; pub use error::RibStopped; -pub use event::RibEvent; pub use local_rib::{LocalRib, LocalRibEvent, LocalRoute, LocalRouteAttrs}; +pub use peer::PeerId; diff --git a/crates/rib/src/local_rib.rs b/crates/rib/src/local_rib.rs index c7dd3e5..7979e62 100644 --- a/crates/rib/src/local_rib.rs +++ b/crates/rib/src/local_rib.rs @@ -146,7 +146,9 @@ impl LocalRib { answer.await.map_err(|_| STOPPED) } - /// Resolves when the actor has stopped. See [`crate::AdjRibIn::stopped`]. + /// Resolves when the actor has stopped. It never does in a healthy + /// process; whoever owns the RIB waits on this to notice a failure even + /// when nothing happens to be using the table. pub async fn stopped(&self) { self.tx.closed().await } @@ -158,8 +160,8 @@ impl LocalRib { self.actor.abort(); } - /// Subscribe to mutations. See [`crate::AdjRibIn::subscribe`] for lag - /// semantics. + /// Subscribe to mutations. A subscriber that falls too far behind is + /// told so with [`broadcast::error::RecvError::Lagged`]. pub fn subscribe(&self) -> broadcast::Receiver { self.events.subscribe() } diff --git a/crates/rib/src/peer.rs b/crates/rib/src/peer.rs new file mode 100644 index 0000000..808d3aa --- /dev/null +++ b/crates/rib/src/peer.rs @@ -0,0 +1,14 @@ +use std::net::SocketAddr; + +/// Identifies a BGP peer by its TCP address and negotiated ASN. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct PeerId { + pub addr: SocketAddr, + pub asn: u32, +} + +impl std::fmt::Display for PeerId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}(AS{})", self.addr, self.asn) + } +} diff --git a/crates/rib/src/table/attrs.rs b/crates/rib/src/table/attrs.rs index 88c11bd..db2d8b3 100644 --- a/crates/rib/src/table/attrs.rs +++ b/crates/rib/src/table/attrs.rs @@ -321,6 +321,24 @@ impl AttrStore { Some(set) } + /// The set for attributes as a session received them: `wire` is what + /// the decoder kept of the UPDATE's attributes + /// (`DecodedUpdate::attributes_wire`), and `next_hop` the next hop of + /// the routes if they came in MP_REACH_NLRI. + pub fn intern_received( + &self, + wire: &[u8], + next_hop: Option<&NlriNextHop>, + ) -> Option> { + let Some(next_hop) = next_hop else { + return self.intern_wire(wire); + }; + let mut with_next_hop = BytesMut::with_capacity(wire.len() + 36); + with_next_hop.put_slice(wire); + encode(&[], Some(next_hop), &mut with_next_hop).ok()?; + self.intern_wire(&with_next_hop) + } + /// How many distinct sets are stored. pub fn len(&self) -> usize { (0..STRIPES).map(|i| self.stripe_at(i).len()).sum() diff --git a/crates/rib/src/table/journal.rs b/crates/rib/src/table/journal.rs index 0dea3d6..6e9310c 100644 --- a/crates/rib/src/table/journal.rs +++ b/crates/rib/src/table/journal.rs @@ -275,6 +275,25 @@ impl RibTable { /// A new follower of this table's best routes. It is given the whole /// table first and every change after. pub fn reader(self: &Arc) -> Reader { + self.register(Phase::Resync { + next: 0, + announced: false, + }) + } + + /// A new follower that is given only the changes from now on, for + /// those who want the news and not the table. If it falls too far + /// behind it is reset like any other, and is then given the table. + pub fn reader_from_now(self: &Arc) -> Reader { + let mut reader = self.register(Phase::Follow { next: 0 }); + for (index, shard) in self.shards.iter().enumerate() { + let shard = read(shard); + reader.set_cursor(index, &shard, shard.journal.seq); + } + reader + } + + fn register(self: &Arc, phase: Phase) -> Reader { let mut slots = self .reader_slots .lock() @@ -300,10 +319,7 @@ impl RibTable { table: Arc::clone(self), slot, cursors: vec![NO_CURSOR; self.shards.len()], - phase: Phase::Resync { - next: 0, - announced: false, - }, + phase, seen: 0, caught_up: false, } diff --git a/crates/rib/src/table/journal_tests.rs b/crates/rib/src/table/journal_tests.rs index 0c1013a..716ffdf 100644 --- a/crates/rib/src/table/journal_tests.rs +++ b/crates/rib/src/table/journal_tests.rs @@ -124,6 +124,31 @@ fn a_new_reader_is_given_the_table_and_then_the_changes() { mirror.assert_matches(&fixture.table); } +#[test] +fn a_reader_from_now_is_given_the_changes_and_not_the_table() { + let fixture = Fixture::new(&[external(1)]); + let (old, new) = (prefix("10.0.0.0/8"), prefix("10.1.0.0/16")); + fixture.announce(0, &old, &Route::via(&[1])); + + let mut reader = fixture.table.reader_from_now(); + assert_eq!( + reader + .poll(100, |event| panic!("unexpected {event:?}")) + .events, + 0 + ); + + fixture.announce(0, &new, &Route::via(&[1])); + fixture.withdraw(0, &old); + let (mut best, mut withdrawn) = (Vec::new(), Vec::new()); + reader.poll(100, |event| match event { + Event::Best { prefix, .. } => best.push(prefix), + Event::Withdrawn { prefix } => withdrawn.push(prefix), + other => panic!("unexpected {other:?}"), + }); + assert_eq!((best, withdrawn), (vec![new], vec![old])); +} + /// A follower wants what the table has now, not how it got there. #[test] fn many_changes_to_a_prefix_between_reads_are_one_event() { diff --git a/crates/rib/src/table/mod.rs b/crates/rib/src/table/mod.rs index 30c0f71..5b44a1d 100644 --- a/crates/rib/src/table/mod.rs +++ b/crates/rib/src/table/mod.rs @@ -401,6 +401,29 @@ impl RibTable { } } + /// Up to `limit` routes, with whether each is its prefix's best: every + /// route of as many prefixes as fit, in no particular order. For showing + /// some of a table that is too large to show all of. + pub fn some_routes(&self, limit: usize) -> Vec<(IpPrefix, PeerIndex, Arc, bool)> { + let mut routes = Vec::new(); + for shard in &self.shards { + for row in &read(shard).rows { + for (i, path) in row.paths.iter().enumerate() { + if routes.len() == limit { + return routes; + } + routes.push(( + row.prefix, + path.peer(), + Arc::clone(&path.attrs), + i == usize::from(row.best), + )); + } + } + } + routes + } + /// Calls `visit` with every route in the table and whether it is its /// prefix's best. As [`RibTable::for_each_best`], without stopping the /// table. diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index 301b8d3..7616c0d 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -13,17 +13,20 @@ use std::pin::Pin; use std::sync::Arc; use bgp_packet::errors::{BgpError, DecodeError, MsgHeaderSubcode}; -use bgp_packet::message::{BGP_HEADER_LEN, BGP_MAX_MSG_LEN, Message, PathAttribute, UpdateMessage}; +use bgp_packet::message::{ + BGP_HEADER_LEN, BGP_MAX_MSG_LEN, Decoded, Message, PathAttribute, UpdateMessage, +}; use bgp_packet::update_decode::UpdateError; use bgp_primitives::parser::ParserContext; -use bgp_rib::{AdjRibIn, PeerId, ReceivedUpdate, RibStopped}; +use bgp_rib::PeerId; +use bgp_rib::table::{Outcome, PeerIndex, PeerInfo, RibTable, TooManyPeers}; use bytes::BytesMut; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf}; use tokio::sync::mpsc; use tokio::time::{Instant, Interval, Sleep, interval_at, sleep}; -use tracing::{error, info, warn}; +use tracing::{debug, error, info, warn}; use crate::collision::{CollisionTable, ConnectionGuard, Direction}; use crate::error::SessionError; @@ -33,6 +36,7 @@ use crate::metrics::{ MESSAGES_RECEIVED, MESSAGES_SENT, SESSIONS_ESTABLISHED_TOTAL, UPDATE_ERRORS, message_type_name, }; use crate::panic::contained; +use crate::routes; use crate::session::{SessionGuard, SessionHandle, SessionRegistry}; /// Number of outbound messages that may be queued for a session before @@ -45,7 +49,8 @@ pub struct SessionConfig { pub router_id: Ipv4Addr, pub local_asn: u32, pub hold_time_secs: u16, - pub adj_rib_in: AdjRibIn, + /// Where routes learned from the peer go. + pub rib: Arc, pub registry: SessionRegistry, /// Shared table used to resolve connection collisions (RFC 4271 §6.8). pub collisions: CollisionTable, @@ -85,8 +90,8 @@ pub async fn run_session( }; // Whatever ended the session, routes learned over it are gone - // (RFC 4271 §8.2.2). Do this before the registration drops so - // subscribers see the withdrawals before the Down event. + // (RFC 4271 §8.2.2). Do this before the registration drops so that the + // routes are withdrawn before the session is seen to be down. if let Some(established) = session.established.take() { let Established { peer_id, @@ -94,11 +99,13 @@ pub async fn run_session( _registration, .. } = established; - match routes.remove().await { - Ok(removed) => info!(peer = %peer_id, removed_routes = removed, "Session ended"), - // No table left to remove them from. - Err(e) => error!(peer = %peer_id, err = %e, "Session ended"), - } + let outcome = routes.remove().await; + info!( + peer = %peer_id, + removed_routes = outcome.removed, + best_changed = outcome.best_changed, + "Session ended" + ); } result } @@ -129,6 +136,7 @@ impl Session { keepalive_timer: None, collision: None, established: None, + attributes_wire: None, peer_addr, local_addr, config, @@ -140,6 +148,8 @@ impl Session { /// What exists only once a session is Established. struct Established { peer_id: PeerId, + /// The peer's number in the routing table. + rib_peer: PeerIndex, /// Messages the rest of the server queued for the peer. outbound: mpsc::Receiver, /// The routes learned from the peer, which go when the session does. @@ -157,24 +167,36 @@ struct Established { /// the task being aborted), dropping this removes them all the same. Left /// in the table they would be offered as routes of a peer that is gone. struct PeerRoutes { - adj_rib_in: AdjRibIn, - peer: PeerId, + table: Arc, + peer: PeerIndex, removed: bool, } impl PeerRoutes { - fn new(adj_rib_in: AdjRibIn, peer: PeerId) -> Self { + fn new(table: Arc, peer: PeerIndex) -> Self { Self { - adj_rib_in, + table, peer, removed: false, } } - /// Removes the routes and returns how many there were. - async fn remove(mut self) -> Result { + /// Removes the routes and says what that did. + async fn remove(mut self) -> Outcome { self.removed = true; - self.adj_rib_in.remove_peer(self.peer).await + let (table, peer) = (Arc::clone(&self.table), self.peer); + // Looks at every prefix, which for a full table is a tenth of a + // second: too long to hold up the other sessions on this thread. + tokio::task::spawn_blocking(move || Self::remove_now(&table, peer)) + .await + .unwrap_or_default() + } + + fn remove_now(table: &RibTable, peer: PeerIndex) -> Outcome { + let outcome = table.remove_peer(peer); + // Most of the peer's attribute sets were its alone. + table.attrs().sweep(); + outcome } } @@ -184,20 +206,22 @@ impl Drop for PeerRoutes { return; } // Not reached in the normal course of things. Drop cannot wait, so - // the removal is handed to a task, if there is still a runtime to - // run it. - let (adj_rib_in, peer) = (self.adj_rib_in.clone(), self.peer); - if let Ok(runtime) = tokio::runtime::Handle::try_current() { - runtime.spawn(async move { - match adj_rib_in.remove_peer(peer).await { - Ok(removed) => warn!( - %peer, - removed_routes = removed, + // the removal is handed to a thread that can take its time, if + // there is still a runtime to give it to. + let (table, peer) = (Arc::clone(&self.table), self.peer); + match tokio::runtime::Handle::try_current() { + Ok(runtime) => { + runtime.spawn_blocking(move || { + let outcome = Self::remove_now(&table, peer); + warn!( + removed_routes = outcome.removed, "Session task ended abnormally; removed its routes" - ), - Err(e) => error!(%peer, err = %e, "Could not remove routes of ended session"), - } - }); + ); + }); + } + Err(_) => { + Self::remove_now(&table, peer); + } } } } @@ -214,6 +238,10 @@ struct Session { /// This connection's slot in the collision table, once admitted. collision: Option, established: Option, + /// The attributes of the UPDATE that is being acted on, as they were + /// on the wire, if the decoder kept them. Set when the message is read + /// and taken when its routes are stored. + attributes_wire: Option>, peer_addr: SocketAddr, local_addr: SocketAddr, config: SessionConfig, @@ -255,9 +283,10 @@ impl Session { async fn next_event(&mut self) -> Result { tokio::select! { frame = self.frames.next(&self.ctx) => match frame { - Ok((msg, update_errors)) => { - self.report(&update_errors); - Ok(Event::Message(msg)) + Ok(decoded) => { + self.report(&decoded.update_errors); + self.attributes_wire = decoded.attributes_wire; + Ok(Event::Message(decoded.message)) } Err(SessionError::Decode(e)) => Ok(Event::DecodeFailed(e)), Err(e) => Err(e), @@ -342,7 +371,13 @@ impl Session { } } } - Action::Established(negotiated) => self.establish(negotiated), + Action::Established(negotiated) => { + if let Err(full) = self.establish(negotiated) { + // What the machine wanted next no longer applies. + error!(peer_addr = %self.peer_addr, err = %full, "Cannot take the peer's routes"); + queue = self.fsm.handle(Event::TableFull(full)).into(); + } + } Action::RoutesIgnored(count) => { warn!( peer_addr = %self.peer_addr, @@ -351,18 +386,26 @@ impl Session { ); } Action::DeliverUpdate(update) => { + let attributes_wire = self.attributes_wire.take(); if let Some(established) = &self.established { - let (withdrawn, announced) = prefix_counts(&update); - info!(peer = %established.peer_id, withdrawn, announced, "Received UPDATE"); - let applied = self - .config - .adj_rib_in - .apply(established.peer_id, ReceivedUpdate::from_message(&update)) - .await; - if let Err(e) = applied { - // What the machine wanted next no longer applies. - error!(peer = %established.peer_id, err = %e, "Cannot store routes"); - queue = self.fsm.handle(Event::RibStopped(e)).into(); + let applied = routes::apply( + &self.config.rib, + established.rib_peer, + &update, + attributes_wire.as_deref(), + ); + match applied { + // One line for every UPDATE is too much to log + // by default: a full table is a million of them. + Ok(outcome) => debug!( + peer = %established.peer_id, + added = outcome.added, + replaced = outcome.replaced, + removed = outcome.removed, + best_changed = outcome.best_changed, + "Received UPDATE" + ), + Err(e) => return Some(Err(e.into())), } } } @@ -378,12 +421,19 @@ impl Session { None } - /// Makes the session visible to the rest of the server. - fn establish(&mut self, negotiated: Negotiated) { + /// Makes room for the peer's routes and makes the session visible to + /// the rest of the server. + fn establish(&mut self, negotiated: Negotiated) -> Result<(), TooManyPeers> { let peer_id = PeerId { addr: self.peer_addr, asn: negotiated.peer_asn, }; + let rib_peer = self.config.rib.add_peer(PeerInfo { + asn: negotiated.peer_asn, + router_id: negotiated.peer_identifier, + address: self.peer_addr.ip(), + ibgp: negotiated.peer_asn == self.config.local_asn, + })?; info!(peer_addr = %self.peer_addr, peer_asn = negotiated.peer_asn, "Session established"); SESSIONS_ESTABLISHED_TOTAL.inc(); @@ -396,22 +446,24 @@ impl Session { self.config.peer_name.clone(), self.config.direction, self.local_addr, - negotiated.four_octet_asn, - negotiated.families, + &negotiated, + rib_peer, outbound_tx, )); self.established = Some(Established { peer_id, + rib_peer, outbound, - routes: PeerRoutes::new(self.config.adj_rib_in.clone(), peer_id), + routes: PeerRoutes::new(Arc::clone(&self.config.rib), rib_peer), _registration: registration, }); + Ok(()) } async fn send(&mut self, msg: &Message) -> Result<(), SessionError> { if let (Message::Update(update), Some(established)) = (msg, &self.established) { let (withdrawn, announced) = prefix_counts(update); - info!(peer = %established.peer_id, withdrawn, announced, "Sending UPDATE"); + debug!(peer = %established.peer_id, withdrawn, announced, "Sending UPDATE"); } if let Message::Notification(notification) = msg { info!(peer_addr = %self.peer_addr, error = %notification.error, "Sending NOTIFICATION, closing session"); @@ -500,20 +552,17 @@ impl FrameReader { /// With an UPDATE come the errors in it that were handled without /// resetting the session (RFC 7606); the UPDATE returned is the one to /// act on, with their handling already applied. - async fn next( - &mut self, - ctx: &ParserContext, - ) -> Result<(Message, Vec), SessionError> { + async fn next(&mut self, ctx: &ParserContext) -> Result { loop { if let Some(length) = frame_length(&self.buf)? && self.buf.len() >= length { let frame = self.buf.split_to(length); - let (msg, update_errors) = Message::decode_with_report(ctx, &frame)?; + let decoded = Message::decode_with_report(ctx, &frame)?; MESSAGES_RECEIVED - .with_label_values(&[message_type_name(&msg)]) + .with_label_values(&[message_type_name(&decoded.message)]) .inc(); - return Ok((msg, update_errors)); + return Ok(decoded); } if self.reader.read_buf(&mut self.buf).await? == 0 { return Err(io::Error::new( @@ -592,16 +641,25 @@ mod tests { peer.write_all(&bytes[..47]).await.unwrap(); assert!(matches!( frames.next(&ctx).await, - Ok((Message::KeepAlive, _)) + Ok(Decoded { + message: Message::KeepAlive, + .. + }) )); assert!(matches!( frames.next(&ctx).await, - Ok((Message::KeepAlive, _)) + Ok(Decoded { + message: Message::KeepAlive, + .. + }) )); peer.write_all(&bytes[47..]).await.unwrap(); assert!(matches!( frames.next(&ctx).await, - Ok((Message::KeepAlive, _)) + Ok(Decoded { + message: Message::KeepAlive, + .. + }) )); drop(peer); @@ -628,7 +686,10 @@ mod tests { peer.write_all(&KEEPALIVE[10..]).await.unwrap(); assert!(matches!( frames.next(&ctx).await, - Ok((Message::KeepAlive, _)) + Ok(Decoded { + message: Message::KeepAlive, + .. + }) )); } @@ -645,7 +706,7 @@ mod tests { router_id: Ipv4Addr::new(10, 0, 0, 1), local_asn: 65000, hold_time_secs: 90, - adj_rib_in: AdjRibIn::spawn(), + rib: Arc::new(RibTable::new(4)), registry: SessionRegistry::new(), collisions: CollisionTable::new(false), direction: Direction::Inbound, @@ -676,11 +737,15 @@ mod tests { /// not leave the peer's routes behind. #[tokio::test] async fn routes_are_removed_even_if_the_session_never_cleans_up() { - let adj_rib_in = AdjRibIn::spawn(); - let peer = PeerId { - addr: ([192, 0, 2, 1], 179).into(), - asn: 65001, - }; + let table = Arc::new(RibTable::new(4)); + let peer = table + .add_peer(PeerInfo { + asn: 65001, + router_id: Ipv4Addr::new(10, 0, 0, 2), + address: Ipv4Addr::new(192, 0, 2, 1).into(), + ibgp: false, + }) + .unwrap(); let update = UpdateMessage { withdrawn_routes: vec![], path_attributes: vec![], @@ -688,22 +753,20 @@ mod tests { bgp_primitives::ip_prefix::IpPrefix::try_from("10.0.0.0/8").unwrap(), ], }; - adj_rib_in - .apply(peer, ReceivedUpdate::from_message(&update)) - .await - .unwrap(); - assert_eq!(adj_rib_in.len().await, Ok(1)); + routes::apply(&table, peer, &update, None).unwrap(); + assert_eq!(table.len(), 1); + assert_eq!(table.attrs().len(), 1); // As when a panic unwinds the session's task. - drop(PeerRoutes::new(adj_rib_in.clone(), peer)); + drop(PeerRoutes::new(Arc::clone(&table), peer)); tokio::time::timeout(Duration::from_secs(5), async { - while adj_rib_in.len().await != Ok(0) { + while !table.is_empty() || !table.attrs().is_empty() { tokio::task::yield_now().await; } }) .await - .expect("the routes are removed"); + .expect("the routes and their attributes are removed"); } proptest! { @@ -751,7 +814,7 @@ mod tests { }; let reader = async { let mut received = Vec::new(); - while let Ok((msg, _)) = frames.next(&ctx).await { + while let Ok(Decoded { message: msg, .. }) = frames.next(&ctx).await { received.push(msg); } received diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs index 8456930..7f2547d 100644 --- a/crates/server/src/error.rs +++ b/crates/server/src/error.rs @@ -7,6 +7,7 @@ use bgp_packet::errors::{BgpError, CeaseSubcode, DecodeError, FsmSubcode, OpenMs use bgp_packet::message::NotificationMessage; use bgp_primitives::parser::EncodeError; use bgp_rib::RibStopped; +use bgp_rib::table::TooManyPeers; use thiserror::Error; use crate::family::FamilySet; @@ -42,10 +43,9 @@ pub enum SessionError { Collision, /// The peer offers none of the address families configured for it, so /// the session could carry nothing. - /// The table the session's routes go into is gone. The server is on - /// its way down; the peer is told so. + /// The routing table has no room for another peer. #[error(transparent)] - RibStopped(#[from] RibStopped), + TableFull(#[from] TooManyPeers), /// A bug: the session's own code panicked. The peer is told the session /// is being reset, and the session is cleaned up like any other. #[error("the session panicked: {0}")] @@ -77,7 +77,7 @@ impl SessionError { Self::OpenRefused(error) => error.clone(), Self::WrongPeerAs { .. } => BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), Self::Collision => BgpError::Cease(CeaseSubcode::ConnCollisionResolution), - Self::RibStopped(_) => BgpError::Cease(CeaseSubcode::AdminShutdown), + Self::TableFull(_) => BgpError::Cease(CeaseSubcode::OutOfResources), // The session will be set up again; nothing is wrong at the // peer's end. Self::Panicked(_) => BgpError::Cease(CeaseSubcode::AdminReset), diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index a80d1cf..2870a09 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -18,7 +18,7 @@ use bgp_packet::errors::{BgpError, DecodeError, OpenMsgSubcode}; use bgp_packet::message::{AS_TRANS, BGP4_VERSION, Message, UpdateMessage}; use bgp_packet::open::{Capability, OpenMessage, OpenOption}; -use bgp_rib::RibStopped; +use bgp_rib::table::TooManyPeers; use crate::error::SessionError; use crate::family::FamilySet; @@ -94,9 +94,9 @@ pub enum Event { CollisionLost, /// The rest of the server wants this message sent to the peer. Outbound(Message), - /// The table that routes from the peer go into has stopped, so the - /// session cannot go on. - RibStopped(RibStopped), + /// The routing table has no room for this peer's routes, so the session + /// cannot go on. + TableFull(TooManyPeers), } /// Something the driver must do, in the order given. @@ -185,7 +185,7 @@ impl Fsm { Event::Message(msg) => self.on_message(msg), Event::DecodeFailed(e) => self.fail(SessionError::Decode(e)), Event::CollisionLost => self.fail(SessionError::Collision), - Event::RibStopped(e) => self.fail(SessionError::RibStopped(e)), + Event::TableFull(e) => self.fail(SessionError::TableFull(e)), Event::HoldTimerExpired => self.fail(match self.phase { Phase::Established(_) => SessionError::HoldTimerExpired, _ => SessionError::Timeout(self.state()), @@ -842,13 +842,12 @@ mod tests { } #[test] - fn losing_the_rib_ends_the_session_and_tells_the_peer_we_are_going() { + fn a_full_table_ends_the_session_and_says_why() { let mut fsm = fsm_in(State::Established, 90); - let stopped = RibStopped { rib: "Adj-RIB-In" }; - let actions = fsm.handle(Event::RibStopped(stopped)); + let actions = fsm.handle(Event::TableFull(TooManyPeers)); let (sent, result) = ending(&fsm, actions); - assert_eq!(sent, Some(BgpError::Cease(CeaseSubcode::AdminShutdown))); - assert!(matches!(result, Err(SessionError::RibStopped(e)) if e == stopped)); + assert_eq!(sent, Some(BgpError::Cease(CeaseSubcode::OutOfResources))); + assert!(matches!(result, Err(SessionError::TableFull(_)))); } #[test] diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 350f958..ca003cc 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -8,6 +8,7 @@ pub mod family; pub mod fsm; pub mod metrics; mod panic; +mod routes; pub mod server; pub mod session; mod sync; diff --git a/crates/server/src/metrics.rs b/crates/server/src/metrics.rs index 7fb3182..ad0dcb9 100644 --- a/crates/server/src/metrics.rs +++ b/crates/server/src/metrics.rs @@ -65,10 +65,34 @@ pub static LOCAL_ROUTES: LazyLock = LazyLock::new(|| { .expect("metric registration") }); -pub static ADJ_RIB_IN_ROUTES: LazyLock = LazyLock::new(|| { +pub static RIB_ROUTES: LazyLock = LazyLock::new(|| { register_int_gauge!( - "bgp_adj_rib_in_routes", - "Routes held in Adj-RIB-In across all peers" + "bgp_rib_routes", + "Routes learned from peers, counting each peer's separately" + ) + .expect("metric registration") +}); + +pub static RIB_PREFIXES: LazyLock = LazyLock::new(|| { + register_int_gauge!( + "bgp_rib_prefixes", + "Prefixes with at least one route, and so with a best route" + ) + .expect("metric registration") +}); + +pub static RIB_ATTRIBUTE_SETS: LazyLock = LazyLock::new(|| { + register_int_gauge!( + "bgp_rib_attribute_sets", + "Distinct sets of path attributes held for the routes" + ) + .expect("metric registration") +}); + +pub static RIB_JOURNAL_ENTRIES: LazyLock = LazyLock::new(|| { + register_int_gauge!( + "bgp_rib_journal_entries", + "Changes of best route waiting for a follower to read them" ) .expect("metric registration") }); @@ -83,7 +107,10 @@ pub fn init() { LazyLock::force(&SESSIONS_ESTABLISHED); LazyLock::force(&COLLISIONS); LazyLock::force(&LOCAL_ROUTES); - LazyLock::force(&ADJ_RIB_IN_ROUTES); + LazyLock::force(&RIB_ROUTES); + LazyLock::force(&RIB_PREFIXES); + LazyLock::force(&RIB_ATTRIBUTE_SETS); + LazyLock::force(&RIB_JOURNAL_ENTRIES); } pub fn message_type_name(msg: &Message) -> &'static str { diff --git a/crates/server/src/routes.rs b/crates/server/src/routes.rs new file mode 100644 index 0000000..afdc4d1 --- /dev/null +++ b/crates/server/src/routes.rs @@ -0,0 +1,62 @@ +//! Putting the routes of a received UPDATE into the routing table. + +use std::sync::Arc; + +use bgp_packet::message::{NlriNextHop, PathAttribute, UpdateMessage}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_primitives::parser::EncodeError; +use bgp_rib::table::{AttrSet, Outcome, PeerIndex, RibTable}; + +/// Applies `update`, received from `peer`, to `table`: its withdrawals +/// first and then its announcements, so that a prefix named in both ends up +/// announced (RFC 4271 §4.3). +/// +/// An UPDATE names prefixes in four places: withdrawn routes and NLRI in the +/// message body, which are IPv4, and the MP_UNREACH_NLRI and MP_REACH_NLRI +/// attributes for every other family (RFC 4760). +/// +/// `attributes_wire` is the attributes as the decoder kept them, if it did +/// (see `DecodedUpdate::attributes_wire`). With it the table finds the +/// attribute set from those bytes, without writing the attributes out again. +pub(crate) fn apply( + table: &RibTable, + peer: PeerIndex, + update: &UpdateMessage, + attributes_wire: Option<&[u8]>, +) -> Result { + let mut outcome = Outcome::default(); + + if !update.withdrawn_routes.is_empty() { + outcome += table.withdraw(peer, &update.withdrawn_routes); + } + for attribute in &update.path_attributes { + if let PathAttribute::MpUnreachNlri(mp) = attribute { + outcome += table.withdraw(peer, &mp.prefixes); + } + } + + let attributes = |next_hop: Option<&NlriNextHop>| -> Result, EncodeError> { + if let Some(set) = + attributes_wire.and_then(|wire| table.attrs().intern_received(wire, next_hop)) + { + return Ok(set); + } + table.attrs().intern(&update.path_attributes, next_hop) + }; + let mut announce = |prefixes: &[IpPrefix], next_hop| -> Result<(), EncodeError> { + if !prefixes.is_empty() { + outcome += table.announce(peer, &attributes(next_hop)?, prefixes); + } + Ok(()) + }; + // Routes in the body have their next hop in the NEXT_HOP attribute, + // which is one of the set's attributes; routes in MP_REACH_NLRI bring + // theirs along. + announce(&update.announced_routes, None)?; + for attribute in &update.path_attributes { + if let PathAttribute::MpReachNlri(mp) = attribute { + announce(&mp.prefixes, Some(&mp.next_hop))?; + } + } + Ok(outcome) +} diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index 62f765c..0790ef0 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -9,7 +9,8 @@ use tokio::net::{TcpListener, TcpStream}; use tokio::task::{JoinError, JoinHandle}; use tracing::{error, info, warn}; -use bgp_rib::{AdjRibIn, LocalRib, RibStopped}; +use bgp_rib::table::RibTable; +use bgp_rib::{LocalRib, RibStopped}; use crate::advertise::spawn_advertiser; use crate::collision::{CollisionTable, Direction}; @@ -30,9 +31,10 @@ pub struct BgpServerBuilder { pub struct BgpServer { config: Config, listeners: Vec<(SocketAddr, TcpListener)>, - /// Shared RIB handle — created at bind time so callers can clone it before - /// the accept loop starts (e.g. tests that want to introspect the RIB). - pub adj_rib_in: AdjRibIn, + /// The routing table: every route learned from peers, and the best one + /// for each prefix. Created at bind time so that callers can share it + /// before the accept loop starts. + pub rib: Arc, /// Routes this router originates; announced to every established peer. pub local_rib: LocalRib, /// Live view of established sessions. @@ -65,7 +67,7 @@ impl BgpServerBuilder { } Ok(BgpServer { listeners, - adj_rib_in: AdjRibIn::spawn(), + rib: Arc::new(RibTable::new(table_shards())), local_rib, registry: SessionRegistry::with_configured( self.config @@ -111,7 +113,7 @@ impl BgpServer { router_id: self.config.router_id, local_asn: self.config.local_asn, hold_time_secs: self.config.hold_time_secs, - adj_rib_in: self.adj_rib_in.clone(), + rib: Arc::clone(&self.rib), registry: self.registry.clone(), collisions: CollisionTable::new(self.config.collision_detect_established_state), direction: Direction::Inbound, @@ -148,7 +150,7 @@ impl BgpServer { loop { let (listen_addr, result, listener) = tokio::select! { Some(accepted) = accepting.next() => accepted, - failure = failed(&self.adj_rib_in, &self.local_rib, &mut tasks) => { + failure = failed(&self.local_rib, &mut tasks) => { return Err(give_up(&self.registry, failure).await); } }; @@ -192,6 +194,13 @@ async fn give_up(registry: &SessionRegistry, failure: ServerError) -> ServerErro failure } +/// How many shards the routing table has. Sessions write to it from every +/// core at once; a good many more shards than cores keeps two of them from +/// wanting the same one. +fn table_shards() -> usize { + 16 * std::thread::available_parallelism().map_or(1, |cores| cores.get()) +} + /// How long sessions get to send their Cease when the server gives up. const SHUTDOWN_GRACE: Duration = Duration::from_secs(2); @@ -201,19 +210,14 @@ fn named(name: &'static str, task: JoinHandle<()>) -> NamedTask { Box::pin(async move { (name, task.await) }) } -/// Resolves when something the server depends on has stopped: a RIB's -/// actor or one of `tasks`. In a healthy server it never resolves. +/// Resolves when something the server depends on has stopped: the local +/// RIB's actor or one of `tasks`. In a healthy server it never resolves. /// /// Nothing else would notice. A stopped RIB is only found out by whoever /// next uses it, and a stopped task by nobody at all: the server would go /// on accepting sessions while storing or announcing nothing. -async fn failed( - adj_rib_in: &AdjRibIn, - local_rib: &LocalRib, - tasks: &mut FuturesUnordered, -) -> ServerError { +async fn failed(local_rib: &LocalRib, tasks: &mut FuturesUnordered) -> ServerError { tokio::select! { - () = adj_rib_in.stopped() => RibStopped { rib: "Adj-RIB-In" }.into(), () = local_rib.stopped() => RibStopped { rib: "local RIB" }.into(), Some((task, result)) = tasks.next() => ServerError::TaskStopped { task, @@ -244,7 +248,6 @@ mod tests { /// If one of them panics, nothing else would ever find out. #[tokio::test] async fn a_task_that_panics_is_a_failure_of_the_server() { - let adj_rib_in = AdjRibIn::spawn(); let local_rib = LocalRib::spawn(); let mut tasks: FuturesUnordered = FuturesUnordered::new(); tasks.push(named("dialer", tokio::spawn(std::future::pending()))); @@ -253,7 +256,7 @@ mod tests { tokio::spawn(async { panic!("no route to announce") }), )); - let failure = failed(&adj_rib_in, &local_rib, &mut tasks).await; + let failure = failed(&local_rib, &mut tasks).await; assert_eq!( failure.to_string(), "the advertiser task stopped unexpectedly: it panicked: no route to announce" @@ -262,13 +265,12 @@ mod tests { #[tokio::test] async fn a_stopped_rib_is_a_failure_of_the_server() { - let adj_rib_in = AdjRibIn::spawn(); let local_rib = LocalRib::spawn(); let mut tasks: FuturesUnordered = FuturesUnordered::new(); tasks.push(named("advertiser", tokio::spawn(std::future::pending()))); local_rib.stop_for_testing(); - let failure = failed(&adj_rib_in, &local_rib, &mut tasks).await; + let failure = failed(&local_rib, &mut tasks).await; assert!( matches!(failure, ServerError::RibStopped(RibStopped { rib }) if rib == "local RIB") ); diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index a5f04ec..c92d722 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -16,10 +16,12 @@ use bgp_packet::message::NotificationMessage; use bgp_packet::message::Message; use bgp_rib::PeerId; +use bgp_rib::table::PeerIndex; use tokio::sync::{broadcast, mpsc}; use crate::collision::Direction; use crate::family::FamilySet; +use crate::fsm::Negotiated; use crate::metrics::SESSIONS_ESTABLISHED; use crate::sync::IgnorePoisonRw; @@ -64,6 +66,9 @@ pub struct SessionHandle { /// The address families negotiated with the peer. Routes of any other /// family are neither sent to it nor accepted from it. pub families: FamilySet, + /// The peer's number in the routing table: what its routes there are + /// marked with. + pub rib_peer: PeerIndex, outbound: mpsc::Sender, } @@ -77,8 +82,8 @@ impl SessionHandle { name: Option, direction: Direction, local_addr: SocketAddr, - four_octet_asn: bool, - families: FamilySet, + negotiated: &Negotiated, + rib_peer: PeerIndex, outbound: mpsc::Sender, ) -> Self { Self { @@ -87,8 +92,9 @@ impl SessionHandle { direction, local_addr, established_at: SystemTime::now(), - four_octet_asn, - families, + four_octet_asn: negotiated.four_octet_asn, + families: negotiated.families.clone(), + rib_peer, outbound, } } diff --git a/crates/web/Cargo.toml b/crates/web/Cargo.toml index 42352ad..051b34b 100644 --- a/crates/web/Cargo.toml +++ b/crates/web/Cargo.toml @@ -11,6 +11,7 @@ version.workspace = true [dependencies] axum.workspace = true bgp-packet.workspace = true +bgp-primitives.workspace = true bgp-rib.workspace = true bgp-server.workspace = true futures.workspace = true diff --git a/crates/web/src/api.rs b/crates/web/src/api.rs index 89c944b..844fefd 100644 --- a/crates/web/src/api.rs +++ b/crates/web/src/api.rs @@ -4,14 +4,18 @@ use std::time::{SystemTime, UNIX_EPOCH}; use axum::Json; -use axum::extract::State; +use std::collections::HashMap; + +use axum::extract::{Query, State}; use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use bgp_packet::community; -use bgp_packet::message::{OriginPathAttribute, PathAttribute}; -use bgp_rib::{LocalRoute, RibEntry, RibStopped, SnapshotEntry}; +use bgp_packet::message::{NlriNextHop, OriginPathAttribute, PathAttribute}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::table::{AttrSet, PeerIndex}; +use bgp_rib::{LocalRoute, PeerId, RibStopped}; use bgp_server::session::{ConfiguredPeer, DialState, SessionHandle, SessionRegistry}; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use crate::AppState; @@ -35,7 +39,10 @@ pub struct StatusView { pub uptime_secs: u64, pub sessions: usize, pub local_routes: usize, + /// Routes learned from peers, counting each peer's separately. pub received_routes: usize, + /// Prefixes with at least one such route, and so with a best route. + pub received_prefixes: usize, } #[derive(Debug, Clone, Serialize)] @@ -187,14 +194,14 @@ pub struct AttrSummary { pub large_communities: Vec, } -pub fn summarize(entry: &RibEntry) -> AttrSummary { +pub fn summarize(attributes: &[PathAttribute], next_hop: Option<&NlriNextHop>) -> AttrSummary { let mut s = AttrSummary { // A multiprotocol route has its next hop here; an IPv4 route has it // in the NEXT_HOP attribute below. - next_hop: entry.next_hop.as_ref().map(ToString::to_string), + next_hop: next_hop.map(ToString::to_string), ..Default::default() }; - for attr in entry.path_attributes.iter() { + for attr in attributes { match attr { PathAttribute::Origin(o) => s.origin = Some(origin_name(*o).to_string()), PathAttribute::AsPath(p) => { @@ -228,23 +235,48 @@ pub struct ReceivedRouteView { pub peer_addr: String, pub peer_asn: u32, pub prefix: String, - pub received_at_unix: u64, + /// Whether this is the route chosen for its prefix. + pub best: bool, pub summary: AttrSummary, /// The decoded attribute list. MP_REACH_NLRI is not in it: its next hop /// is in the summary and its prefixes are routes of their own. pub attributes: Vec, } -impl From<&SnapshotEntry> for ReceivedRouteView { - fn from(e: &SnapshotEntry) -> Self { - Self { - peer_addr: e.peer.addr.to_string(), - peer_asn: e.peer.asn, - prefix: e.prefix.to_string(), - received_at_unix: unix_secs(e.entry.received_at), - summary: summarize(&e.entry), - attributes: e.entry.path_attributes.to_vec(), - } +/// Who the routes in the table are from: the table knows peers by number, +/// the sessions know which number is whose. +pub(crate) struct PeerNames(HashMap); + +impl PeerNames { + pub(crate) fn of(registry: &SessionRegistry) -> Self { + Self( + registry + .list() + .into_iter() + .map(|session| (session.rib_peer, session.peer)) + .collect(), + ) + } + + /// A view of one route. `None` if the route's session has just ended: + /// the route is on its way out of the table. + pub(crate) fn view( + &self, + prefix: &IpPrefix, + peer: PeerIndex, + attrs: &AttrSet, + best: bool, + ) -> Option { + let peer = self.0.get(&peer)?; + let (attributes, next_hop) = attrs.attributes(); + Some(ReceivedRouteView { + peer_addr: peer.addr.to_string(), + peer_asn: peer.asn, + prefix: prefix.to_string(), + best, + summary: summarize(&attributes, next_hop.as_ref()), + attributes, + }) } } @@ -286,7 +318,8 @@ pub async fn status(State(state): State) -> Result, U uptime_secs: state.info.started.elapsed().as_secs(), sessions: state.registry.list().len(), local_routes: state.local_rib.len().await?, - received_routes: state.adj_rib_in.len().await?, + received_routes: state.rib.len(), + received_prefixes: state.rib.prefix_count(), })) } @@ -322,16 +355,29 @@ pub async fn local_routes( Ok(Json(routes)) } +/// How many routes [`received_routes`] returns unless asked for another +/// number. The table can hold hundreds of millions. +const DEFAULT_ROUTE_LIMIT: usize = 1000; + +#[derive(Debug, Deserialize)] +pub struct RouteQuery { + /// At most this many routes. + limit: Option, +} + +/// Some of the routes learned from peers: all the routes of as many +/// prefixes as fit in the limit. `/api/status` has the totals. pub async fn received_routes( State(state): State, -) -> Result>, Unavailable> { + Query(query): Query, +) -> Json> { + let names = PeerNames::of(&state.registry); let mut routes: Vec = state - .adj_rib_in - .snapshot() - .await? + .rib + .some_routes(query.limit.unwrap_or(DEFAULT_ROUTE_LIMIT)) .iter() - .map(ReceivedRouteView::from) + .filter_map(|(prefix, peer, attrs, best)| names.view(prefix, *peer, attrs, *best)) .collect(); routes.sort_by(|a, b| (&a.prefix, &a.peer_addr).cmp(&(&b.prefix, &b.peer_addr))); - Ok(Json(routes)) + Json(routes) } diff --git a/crates/web/src/events.rs b/crates/web/src/events.rs index c1ec555..a81474b 100644 --- a/crates/web/src/events.rs +++ b/crates/web/src/events.rs @@ -1,19 +1,23 @@ -//! Server-sent event stream merging session, local-route and received-route +//! Server-sent event stream merging session, local-route and best-route //! changes into one feed the UI can follow. use std::convert::Infallible; +use std::sync::Arc; use axum::extract::State; use axum::response::sse::{Event, KeepAlive, Sse}; -use bgp_rib::{LocalRibEvent, RibEvent}; +use bgp_primitives::ip_prefix::IpPrefix; +use bgp_rib::LocalRibEvent; +use bgp_rib::table::{self as journal, AttrSet, PeerIndex}; use bgp_server::session::SessionEvent; use futures::stream::{Stream, StreamExt}; use serde::Serialize; -use tokio_stream::wrappers::BroadcastStream; +use tokio::sync::mpsc; use tokio_stream::wrappers::errors::BroadcastStreamRecvError; +use tokio_stream::wrappers::{BroadcastStream, ReceiverStream}; use crate::AppState; -use crate::api::{LocalRouteView, PeerView, ReceivedRouteView, summarize}; +use crate::api::{LocalRouteView, PeerNames, PeerView, ReceivedRouteView}; #[derive(Debug, Serialize)] #[serde(tag = "type", rename_all = "snake_case")] @@ -35,16 +39,17 @@ pub enum WireEvent { LocalWithdrawn { prefix: String, }, - ReceivedAnnounced { + /// The best route for a prefix is now this one. + BestChanged { route: ReceivedRouteView, }, - ReceivedWithdrawn { - peer_addr: String, - peer_asn: u32, + /// A prefix has no route any more. + BestWithdrawn { prefix: String, }, - /// The client fell behind and missed `missed` events on `stream`; it - /// should refetch the corresponding table. + /// The client fell behind and missed events on `stream` (`missed` of + /// them, where that is known); it should refetch the corresponding + /// table. Lagged { stream: &'static str, missed: u64, @@ -85,38 +90,88 @@ impl WireEvent { }, } } +} - fn from_received(ev: Result) -> Self { - match ev { - Ok(RibEvent::Announced { - peer, - prefix, - entry, - }) => Self::ReceivedAnnounced { - route: ReceivedRouteView { - peer_addr: peer.addr.to_string(), - peer_asn: peer.asn, - prefix: prefix.to_string(), - received_at_unix: entry - .received_at - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or(0), - summary: summarize(&entry), - attributes: entry.path_attributes.to_vec(), - }, - }, - Ok(RibEvent::Withdrawn { peer, prefix }) => Self::ReceivedWithdrawn { - peer_addr: peer.addr.to_string(), - peer_asn: peer.asn, - prefix: prefix.to_string(), - }, - Err(BroadcastStreamRecvError::Lagged(n)) => Self::Lagged { - stream: "received_routes", - missed: n, - }, +/// The table's changes of best route as events, from when the client +/// connects. Each client follows the table's journal with a reader of its +/// own, so one that reads slowly holds nobody else up. +fn best_routes(state: &AppState) -> impl Stream + use<> { + let (events, stream) = mpsc::channel(BEST_ROUTE_BUFFER); + let registry = state.registry.clone(); + // Taken before this returns, so that nothing that happens after the + // client is answered is missed. + let mut reader = state.rib.reader_from_now(); + tokio::spawn(async move { + // If the reader falls too far behind it is given the whole table + // again. That is not news: the client is told it missed something + // and gets its table from the API. + let mut syncing = false; + loop { + tokio::select! { + () = reader.changed() => {} + () = events.closed() => return, + } + loop { + // Only what is needed is taken while the table is held; + // the views are made afterwards. + let mut changes = Vec::new(); + let progress = reader.poll(BEST_ROUTE_BUFFER, |event| match event { + journal::Event::Reset => { + syncing = true; + changes.push(Change::Lagged); + } + journal::Event::Synced => syncing = false, + journal::Event::Best { + prefix, + peer, + attrs, + } if !syncing => { + changes.push(Change::Best(prefix, peer, Arc::clone(attrs))); + } + journal::Event::Withdrawn { prefix } if !syncing => { + changes.push(Change::Withdrawn(prefix)); + } + _ => {} + }); + let names = PeerNames::of(®istry); + for change in changes { + let event = match change { + Change::Best(prefix, peer, attrs) => { + match names.view(&prefix, peer, &attrs, true) { + Some(route) => WireEvent::BestChanged { route }, + None => continue, + } + } + Change::Withdrawn(prefix) => WireEvent::BestWithdrawn { + prefix: prefix.to_string(), + }, + Change::Lagged => WireEvent::Lagged { + stream: "best_routes", + missed: 0, + }, + }; + if events.send(event).await.is_err() { + return; + } + } + if progress.caught_up { + break; + } + } } - } + }); + ReceiverStream::new(stream) +} + +/// How many best-route events are read from the journal, and queued for a +/// client, at a time. +const BEST_ROUTE_BUFFER: usize = 256; + +enum Change { + Best(IpPrefix, PeerIndex, Arc), + Withdrawn(IpPrefix), + /// The client's reader fell too far behind the table. + Lagged, } pub async fn stream( @@ -124,7 +179,7 @@ pub async fn stream( ) -> Sse>> { let sessions = BroadcastStream::new(state.registry.subscribe()).map(WireEvent::from_session); let local = BroadcastStream::new(state.local_rib.subscribe()).map(WireEvent::from_local); - let received = BroadcastStream::new(state.adj_rib_in.subscribe()).map(WireEvent::from_received); + let received = best_routes(&state); let merged = futures::stream::select(sessions, futures::stream::select(local, received)).map(|ev| { diff --git a/crates/web/src/lib.rs b/crates/web/src/lib.rs index 46d09ff..7004ff8 100644 --- a/crates/web/src/lib.rs +++ b/crates/web/src/lib.rs @@ -17,7 +17,8 @@ use axum::Router; use axum::http::header; use axum::response::{Html, IntoResponse, Redirect}; use axum::routing::get; -use bgp_rib::{AdjRibIn, LocalRib}; +use bgp_rib::LocalRib; +use bgp_rib::table::RibTable; use bgp_server::session::SessionRegistry; use tokio::net::TcpListener; use tracing::info; @@ -78,7 +79,8 @@ pub struct AppState { pub info: Arc, pub registry: SessionRegistry, pub local_rib: LocalRib, - pub adj_rib_in: AdjRibIn, + /// The routing table: routes learned from peers and the best of them. + pub rib: Arc, } /// The status page, embedded so the binary is self-contained. diff --git a/crates/web/src/metrics.rs b/crates/web/src/metrics.rs index bce368e..ec9ae83 100644 --- a/crates/web/src/metrics.rs +++ b/crates/web/src/metrics.rs @@ -4,26 +4,26 @@ use axum::extract::State; use axum::http::{StatusCode, header}; use axum::response::IntoResponse; -use bgp_rib::RibStopped; -use bgp_server::metrics::{ADJ_RIB_IN_ROUTES, LOCAL_ROUTES}; +use bgp_server::metrics::{ + LOCAL_ROUTES, RIB_ATTRIBUTE_SETS, RIB_JOURNAL_ENTRIES, RIB_PREFIXES, RIB_ROUTES, +}; use prometheus::{Encoder, TextEncoder}; use crate::AppState; pub async fn metricz(State(state): State) -> impl IntoResponse { - // A scrape that cannot read the tables fails. Serving the last known - // sizes would hide that the daemon is broken from the one system whose + // A scrape that cannot read a table fails. Serving the last known + // size would hide that the daemon is broken from the one system whose // job is to notice. - let sizes = async { - Ok::<_, RibStopped>((state.local_rib.len().await?, state.adj_rib_in.len().await?)) - }; - match sizes.await { - Ok((local, received)) => { - LOCAL_ROUTES.set(i64::try_from(local).unwrap_or(i64::MAX)); - ADJ_RIB_IN_ROUTES.set(i64::try_from(received).unwrap_or(i64::MAX)); - } + let gauge = |count: usize| i64::try_from(count).unwrap_or(i64::MAX); + match state.local_rib.len().await { + Ok(local) => LOCAL_ROUTES.set(gauge(local)), Err(e) => return (StatusCode::SERVICE_UNAVAILABLE, e.to_string()).into_response(), } + RIB_ROUTES.set(gauge(state.rib.len())); + RIB_PREFIXES.set(gauge(state.rib.prefix_count())); + RIB_ATTRIBUTE_SETS.set(gauge(state.rib.attrs().len())); + RIB_JOURNAL_ENTRIES.set(gauge(state.rib.journal_len())); let mut buf = Vec::new(); match TextEncoder::new().encode(&prometheus::gather(), &mut buf) { diff --git a/crates/web/static/statusz.html b/crates/web/static/statusz.html index c86db4f..a1afdf4 100644 --- a/crates/web/static/statusz.html +++ b/crates/web/static/statusz.html @@ -67,9 +67,9 @@ View: variables, status json, < -

Received routes, Adj-RIB-In (...)

+

Received routes (...)

- +
prefixfrom peerAS pathnext hoporiginMEDlocal prefcommunitieslarge communitiesage
prefixfrom peerAS pathnext hoporiginMEDlocal prefcommunitieslarge communitiesbest
@@ -162,16 +162,19 @@ View: variables, status json, < } function renderReceived() { - const b = $('received-body'); $('received-count').textContent = state.received.length; + const b = $('received-body'), st = state.status; + $('received-count').textContent = st && st.received_routes > state.received.length + ? `showing ${state.received.length} of ${st.received_routes}, for ${st.received_prefixes} prefixes` + : `${state.received.length}${st ? `, for ${st.received_prefixes} prefixes` : ''}`; if (!state.received.length) { setHTML(b, 'No routes received from peers.'); return; } setHTML(b, state.received.map(r => { const s = r.summary; return ` ${esc(r.prefix)}${esc(r.peer_addr)} AS${r.peer_asn}${asPath(s.as_path)} ${s.next_hop ? esc(s.next_hop) : DASH}${s.origin ? esc(s.origin) : DASH}${s.med ?? DASH}${s.local_pref ?? DASH} - ${commList(s.communities, true)}${commList(s.large_communities)}${age(r.received_at_unix)}`; }).join('')); + ${commList(s.communities, true)}${commList(s.large_communities)}${r.best ? 'best' : ''}`; }).join('')); } // ---- data loading ------------------------------------------------------ - async function loadStatus() { state.status = await getJSON('/api/status'); renderStatus(); renderLocal(); } + async function loadStatus() { state.status = await getJSON('/api/status'); renderStatus(); renderLocal(); renderReceived(); } async function loadPeers() { state.peers = await getJSON('/api/peers'); renderPeers(); } async function loadLocal() { state.local = await getJSON('/api/routes/local'); renderLocal(); } async function loadReceived() { state.received = await getJSON('/api/routes/received'); renderReceived(); } @@ -193,8 +196,8 @@ View: variables, status json, < (r.communities.length ? ` [${names(r.communities)}]` : '') + (r.large_communities.length ? ` [${names(r.large_communities)}]` : '') + (r.prepend ? ` prepend ${r.prepend}` : '') + (r.med != null ? ` med ${r.med}` : ''); } case 'local_withdrawn': return `no longer originating ${esc(ev.prefix)}`; - case 'received_announced': return `received ${esc(ev.route.prefix)} from AS${ev.route.peer_asn} path [${ev.route.summary.as_path.join(' ')}]`; - case 'received_withdrawn': return `peer AS${ev.peer_asn} withdrew ${esc(ev.prefix)}`; + case 'best_changed': return `best route for ${esc(ev.route.prefix)} is now from AS${ev.route.peer_asn} path [${ev.route.summary.as_path.join(' ')}]`; + case 'best_withdrawn': return `${esc(ev.prefix)} has no route any more`; case 'lagged': return `missed ${ev.missed} events on ${esc(ev.stream)}`; default: return esc(JSON.stringify(ev)); } diff --git a/fuzz/src/lib.rs b/fuzz/src/lib.rs index d53896c..ed22dd3 100644 --- a/fuzz/src/lib.rs +++ b/fuzz/src/lib.rs @@ -16,9 +16,10 @@ pub fn decode_and_check(four_octet_asn: bool, ibgp: bool, wire: &[u8]) { four_octet_asn: Some(four_octet_asn), ibgp, }; - let Ok((msg, repairs)) = Message::decode_with_report(&ctx, wire) else { + let Ok(decoded) = Message::decode_with_report(&ctx, wire) else { return; }; + let (msg, repairs) = (decoded.message, decoded.update_errors); let _ = format!("{msg:?} {repairs:?}"); let mut rewritten = BytesMut::new(); @@ -36,9 +37,8 @@ pub fn decode_and_check(four_octet_asn: bool, ibgp: bool, wire: &[u8]) { // An UPDATE the decoder had to repair (RFC 7606) is not what was on the // wire, and need not read back as itself. if repairs.is_empty() { - let (again, repairs) = - Message::decode_with_report(&ctx, &rewritten).expect("and read back"); - assert!(repairs.is_empty(), "{repairs:?}"); - assert_eq!(again, msg); + let again = Message::decode_with_report(&ctx, &rewritten).expect("and read back"); + assert!(again.update_errors.is_empty(), "{:?}", again.update_errors); + assert_eq!(again.message, msg); } }