rayhaan/ferrix
0
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use the sharded table in the server: sessions choose best routes

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Rayhaan JaufeerallyandClaude Fable 5.1 committed 2026-10-02 00:31:28 +02:00
1 parent 3b20576971
commit 31aa3a4563
43 files changed
+1213 -1212
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Cargo.lock
Generated
+1
View File
@@ -274,6 +274,7 @@ version = "0.0.2"
dependencies = [
"axum",
"bgp-packet",
"bgp-primitives",
"bgp-rib",
"bgp-server",
"futures",
bin/bgp/src/main.rs
+1 -1
View File
@@ -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 {
crates/bench/benches/full_table.rs
+127 -255
View File
@@ -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::<IpPrefix>()),
("RibEntry", size_of::<RibEntry>()),
(
"(IpPrefix, RibEntry), a table slot",
size_of::<(IpPrefix, RibEntry)>(),
),
("PathAttribute", size_of::<PathAttribute>()),
("SnapshotEntry", size_of::<SnapshotEntry>()),
] {
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::<IpPrefix>(),
size_of::<PathAttribute>()
);
let prefixes = Prefixes::generate(shape);
let per_peer: Vec<Vec<UpdateMessage>> = (1..=peers)
.map(|peer| announcements(&prefixes, shape, peer))
let wire: Vec<Vec<BytesMut>> = (1..=peers)
.map(|peer| encoded(&announcements(&prefixes, shape, peer)))
.collect();
let wire: Vec<Vec<BytesMut>> = 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<Vec<Message>> = 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<Vec<ReceivedUpdate>> = 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(&region))
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<u8> = 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<ReceivedUpdate> = 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<u8> = 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<UpdateMessage>], 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<BytesMut>], routes: usize, threads: usize) {
let region = Region::begin();
let table = RibTable::new(threads * 16);
let peers: Vec<PeerIndex> = (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<u64, u32>, HashMap<u128, u32>)> = Vec::new();
for updates in per_peer {
let mut v4: HashMap<u64, u32> = HashMap::with_capacity(prefixes.ipv4.len());
let mut v6: HashMap<u128, u32> = 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(&region);
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);
}
crates/bench/benches/real_table.rs
+2 -124
View File
@@ -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<Attributes>],
received_at: SystemTime,
) -> Vec<ReceivedUpdate> {
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<Attributes>],
) {
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<ReceivedUpdate>)> = 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(&region);
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
crates/integration-tests/src/test_server.rs
+94 -6
View File
@@ -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<Result<(), ServerError>>,
/// 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<RibTable>,
/// 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<PathAttribute>,
/// The next hop, if the route came in MP_REACH_NLRI.
pub next_hop: Option<NlriNextHop>,
/// 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<RouteChange>,
}
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<RouteChange> {
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<ReceivedRoute> {
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 }))
crates/integration-tests/tests/announce.rs
+6 -6
View File
@@ -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??;
crates/integration-tests/tests/basic_connection.rs
+14 -21
View File
@@ -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",
crates/integration-tests/tests/best_route.rs
+239
View File
@@ -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<TestPeer> {
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<Vec<u32>> {
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<Value> {
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<Value> = 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<IpPrefix> = (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(())
}
crates/integration-tests/tests/collision.rs
+2 -2
View File
@@ -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(())
crates/integration-tests/tests/failure.rs
+24 -44
View File
@@ -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<BgpError> {
.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(())
}
crates/integration-tests/tests/http_api.rs
+1 -1
View File
@@ -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(())
}
crates/integration-tests/tests/interop.rs
+1 -1
View File
@@ -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(())
}
crates/integration-tests/tests/negotiation.rs
+3 -4
View File
@@ -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(())
crates/integration-tests/tests/update_errors.rs
+7 -10
View File
@@ -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<TestPeer> {
}
/// Waits until the server's Adj-RIB-In holds exactly `prefixes`.
async fn wait_for_routes(server: &TestServer, prefixes: &[&str]) -> Result<Vec<SnapshotEntry>> {
async fn wait_for_routes(server: &TestServer, prefixes: &[&str]) -> Result<Vec<ReceivedRoute>> {
let mut expected: Vec<IpPrefix> = 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)))
};
crates/packet/src/message.rs
+21 -6
View File
@@ -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, DecodeError> {
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<UpdateError>), DecodeError> {
pub fn decode_with_report(ctx: &ParserContext, buf: &[u8]) -> Result<Decoded, DecodeError> {
let mut update_errors = Vec::new();
let mut attributes_wire = None;
let header_error = |subcode, data: Vec<u8>, 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<UpdateError>,
/// See [`crate::update_decode::DecodedUpdate::attributes_wire`].
pub attributes_wire: Option<Vec<u8>>,
}
/// A request to send again every route of one address family (RFC 2918).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct RouteRefreshMessage {
crates/packet/src/update_decode.rs
+71 -7
View File
@@ -65,6 +65,15 @@ pub struct DecodedUpdate {
pub update: UpdateMessage,
/// Empty when the UPDATE was well formed.
pub errors: Vec<UpdateError>,
/// 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<Vec<u8>>,
}
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<UpdateError>,
kept_wire: &mut Vec<u8>,
) -> Result<Vec<PathAttribute>, 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.
crates/packet/tests/properties.rs
+8 -7
View File
@@ -454,9 +454,10 @@ fn session_and_message() -> impl Strategy<Value = (Session, Message)> {
/// 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());
}
crates/rib/src/adj_rib_in.rs
-277
View File
@@ -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<usize>,
},
Snapshot {
reply: oneshot::Sender<Vec<SnapshotEntry>>,
},
Len {
reply: oneshot::Sender<usize>,
},
}
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<RibCommand>,
events: broadcast::Sender<RibEvent>,
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<usize, RibStopped> {
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<Vec<SnapshotEntry>, RibStopped> {
self.ask(|reply| RibCommand::Snapshot { reply }).await
}
/// How many routes the RIB holds, without copying them.
pub async fn len(&self) -> Result<usize, RibStopped> {
self.ask(|reply| RibCommand::Len { reply }).await
}
/// Whether the RIB holds no routes.
pub async fn is_empty(&self) -> Result<bool, RibStopped> {
Ok(self.len().await? == 0)
}
/// Sends a command that is answered, and waits for the answer.
async fn ask<T>(
&self,
command: impl FnOnce(oneshot::Sender<T>) -> RibCommand,
) -> Result<T, RibStopped> {
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<RibEvent> {
self.events.subscribe()
}
}
async fn run_actor(mut rx: mpsc::Receiver<RibCommand>, events: broadcast::Sender<RibEvent>) {
// 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<PeerId, HashMap<IpPrefix, RibEntry>> = 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");
}
}
crates/rib/src/entry.rs
-197
View File
@@ -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<NlriNextHop>,
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<IpPrefix>,
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<IpPrefix>) -> 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<IpPrefix> = (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)
]
);
}
}
crates/rib/src/error.rs
+3 -3
View File
@@ -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,
}
crates/rib/src/event.rs
-16
View File
@@ -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,
},
}
crates/rib/src/lib.rs
+5 -6
View File
@@ -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;
crates/rib/src/local_rib.rs
+5 -3
View File
@@ -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<LocalRibEvent> {
self.events.subscribe()
}
crates/rib/src/peer.rs
+14
View File
@@ -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)
}
}
crates/rib/src/table/attrs.rs
+18
View File
@@ -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<Arc<AttrSet>> {
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()
crates/rib/src/table/journal.rs
+20 -4
View File
@@ -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<Self>) -> 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<Self>) -> 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<Self>, 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,
}
crates/rib/src/table/journal_tests.rs
+25
View File
@@ -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() {
crates/rib/src/table/mod.rs
+23
View File
@@ -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<AttrSet>, 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.
crates/server/src/connection.rs
+141 -78
View File
@@ -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<RibTable>,
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<OwnedReadHalf> {
keepalive_timer: None,
collision: None,
established: None,
attributes_wire: None,
peer_addr,
local_addr,
config,
@@ -140,6 +148,8 @@ impl Session<OwnedReadHalf> {
/// 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<Message>,
/// 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<RibTable>,
peer: PeerIndex,
removed: bool,
}
impl PeerRoutes {
fn new(adj_rib_in: AdjRibIn, peer: PeerId) -> Self {
fn new(table: Arc<RibTable>, 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<usize, RibStopped> {
/// 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<R> {
/// This connection's slot in the collision table, once admitted.
collision: Option<ConnectionGuard>,
established: Option<Established>,
/// 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<Vec<u8>>,
peer_addr: SocketAddr,
local_addr: SocketAddr,
config: SessionConfig,
@@ -255,9 +283,10 @@ impl<R: AsyncRead + Unpin> Session<R> {
async fn next_event(&mut self) -> Result<Event, SessionError> {
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<R: AsyncRead + Unpin> Session<R> {
}
}
}
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<R: AsyncRead + Unpin> Session<R> {
);
}
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<R: AsyncRead + Unpin> Session<R> {
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<R: AsyncRead + Unpin> Session<R> {
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<R: AsyncRead + Unpin> FrameReader<R> {
/// 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<UpdateError>), SessionError> {
async fn next(&mut self, ctx: &ParserContext) -> Result<Decoded, SessionError> {
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
crates/server/src/error.rs
+4 -4
View File
@@ -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),
crates/server/src/fsm.rs
+9 -10
View File
@@ -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]
crates/server/src/lib.rs
+1
View File
@@ -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;
crates/server/src/metrics.rs
+31 -4
View File
@@ -65,10 +65,34 @@ pub static LOCAL_ROUTES: LazyLock<IntGauge> = LazyLock::new(|| {
.expect("metric registration")
});
pub static ADJ_RIB_IN_ROUTES: LazyLock<IntGauge> = LazyLock::new(|| {
pub static RIB_ROUTES: LazyLock<IntGauge> = 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<IntGauge> = 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<IntGauge> = 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<IntGauge> = 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 {
crates/server/src/routes.rs
+62
View File
@@ -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<Outcome, EncodeError> {
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<Arc<AttrSet>, 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)
}
crates/server/src/server.rs
+21 -19
View File
@@ -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<RibTable>,
/// 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<NamedTask>,
) -> ServerError {
async fn failed(local_rib: &LocalRib, tasks: &mut FuturesUnordered<NamedTask>) -> 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<NamedTask> = 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<NamedTask> = 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")
);
crates/server/src/session.rs
+10 -4
View File
@@ -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<Message>,
}
@@ -77,8 +82,8 @@ impl SessionHandle {
name: Option<String>,
direction: Direction,
local_addr: SocketAddr,
four_octet_asn: bool,
families: FamilySet,
negotiated: &Negotiated,
rib_peer: PeerIndex,
outbound: mpsc::Sender<Message>,
) -> 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,
}
}
crates/web/Cargo.toml
+1
View File
@@ -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
crates/web/src/api.rs
+71 -25
View File
@@ -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<String>,
}
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<PathAttribute>,
}
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<PeerIndex, PeerId>);
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<ReceivedRouteView> {
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<AppState>) -> Result<Json<StatusView>, 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<usize>,
}
/// 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<AppState>,
) -> Result<Json<Vec<ReceivedRouteView>>, Unavailable> {
Query(query): Query<RouteQuery>,
) -> Json<Vec<ReceivedRouteView>> {
let names = PeerNames::of(&state.registry);
let mut routes: Vec<ReceivedRouteView> = 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)
}
crates/web/src/events.rs
+96 -41
View File
@@ -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<RibEvent, BroadcastStreamRecvError>) -> 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<Item = WireEvent> + 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(&registry);
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<AttrSet>),
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<impl Stream<Item = Result<Event, Infallible>>> {
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| {
crates/web/src/lib.rs
+4 -2
View File
@@ -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<RouterInfo>,
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<RibTable>,
}
/// The status page, embedded so the binary is self-contained.
crates/web/src/metrics.rs
+12 -12
View File
@@ -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<AppState>) -> 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) {
crates/web/static/statusz.html
+10 -7
View File
@@ -67,9 +67,9 @@ View: <a href="/metricz">variables</a>, <a href="/api/status">status json</a>, <
<tbody id="local-body"></tbody>
</table>
<h1 id="received">Received routes, Adj-RIB-In (<span id="received-count">...</span>)</h1>
<h1 id="received">Received routes (<span id="received-count">...</span>)</h1>
<table class="data">
<thead><tr><th>prefix</th><th>from peer</th><th>AS path</th><th>next hop</th><th>origin</th><th>MED</th><th>local pref</th><th>communities</th><th>large communities</th><th>age</th></tr></thead>
<thead><tr><th>prefix</th><th>from peer</th><th>AS path</th><th>next hop</th><th>origin</th><th>MED</th><th>local pref</th><th>communities</th><th>large communities</th><th>best</th></tr></thead>
<tbody id="received-body"></tbody>
</table>
@@ -162,16 +162,19 @@ View: <a href="/metricz">variables</a>, <a href="/api/status">status json</a>, <
}
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, '<tr><td colspan="10" class="dim">No routes received from peers.</td></tr>'); return; }
setHTML(b, state.received.map(r => { const s = r.summary; return `<tr>
<td class="mono">${esc(r.prefix)}</td><td class="mono">${esc(r.peer_addr)} <span class="dim">AS${r.peer_asn}</span></td><td class="mono">${asPath(s.as_path)}</td>
<td class="mono">${s.next_hop ? esc(s.next_hop) : DASH}</td><td>${s.origin ? esc(s.origin) : DASH}</td><td>${s.med ?? DASH}</td><td>${s.local_pref ?? DASH}</td>
<td class="mono">${commList(s.communities, true)}</td><td class="mono">${commList(s.large_communities)}</td><td>${age(r.received_at_unix)}</td></tr>`; }).join(''));
<td class="mono">${commList(s.communities, true)}</td><td class="mono">${commList(s.large_communities)}</td><td>${r.best ? '<span class="ok">best</span>' : ''}</td></tr>`; }).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: <a href="/metricz">variables</a>, <a href="/api/status">status json</a>, <
(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 `<span class="warn">no longer originating</span> ${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} <span class="warn">withdrew</span> ${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 <span class="warn">no route</span> any more`;
case 'lagged': return `<span class="bad">missed ${ev.missed} events</span> on ${esc(ev.stream)}`;
default: return esc(JSON.stringify(ev));
}
fuzz/src/lib.rs
+5 -5
View File
@@ -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);
}
}