diff --git a/Cargo.lock b/Cargo.lock index 97a412e..c3f83b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -262,6 +262,7 @@ dependencies = [ "proptest", "protobuf", "protobuf-codegen", + "socket2", "thiserror 2.0.12", "tokio", "tracing", diff --git a/Cargo.toml b/Cargo.toml index 752b0d3..4c7d5fd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -41,6 +41,7 @@ bytes = "1.10.1" eyre = "0.6.12" nom = "8.0.0" serde = { version = "1.0.219", features = ["derive"] } +socket2 = "0.6" serde_json = "1.0.140" serde_repr = "0.1.20" thiserror = "2.0.12" diff --git a/bin/bgp/src/main.rs b/bin/bgp/src/main.rs index 071934c..a00176f 100644 --- a/bin/bgp/src/main.rs +++ b/bin/bgp/src/main.rs @@ -1,14 +1,13 @@ use std::sync::Arc; -use std::time::Duration; use bgp_server::config::Config; use bgp_server::proto::config::BgpConfig; -use bgp_server::server::BgpServerBuilder; +use bgp_server::server::{BgpServer, close_sessions}; use bgp_server::tracing::TracingArgs; use bgp_web::{AppState, RouterInfo}; use clap::Parser; use eyre::{Context, Result}; -use tracing::{error, info, warn}; +use tracing::{error, info}; #[derive(Parser, Debug)] struct ServerArgs { @@ -31,9 +30,14 @@ async fn main() -> Result<()> { .wrap_err("Failed to parse config file")?; let config = Config::from_proto(proto_config).wrap_err("Invalid configuration")?; - let server = BgpServerBuilder::new(config).bind().await?; + let server = BgpServer::bind(config).await?; if let Some(http_endpoint) = server.config().http_endpoint { + // Bound here, so that an address that cannot be bound stops the + // daemon from starting, as it does for the BGP listeners. + let listener = tokio::net::TcpListener::bind(http_endpoint) + .await + .wrap_err_with(|| format!("Failed to bind HTTP listener on {http_endpoint}"))?; let state = AppState { info: Arc::new( RouterInfo::new( @@ -48,7 +52,7 @@ async fn main() -> Result<()> { rib: Arc::clone(&server.rib), }; tokio::spawn(async move { - if let Err(e) = bgp_web::serve(state, http_endpoint).await { + if let Err(e) = bgp_web::serve_on(state, listener).await { error!(err = %e, "HTTP server exited"); } }); @@ -62,10 +66,7 @@ async fn main() -> Result<()> { result = server.run() => Ok(result?), signal = shutdown_signal() => { info!(signal, "Shutting down, closing BGP sessions"); - let left = registry.shutdown(Duration::from_secs(2)).await; - if left > 0 { - warn!(sessions = left, "Some sessions did not close in time"); - } + close_sessions(®istry).await; Ok(()) } } diff --git a/crates/integration-tests/src/test_server.rs b/crates/integration-tests/src/test_server.rs index a77b987..5446cb0 100644 --- a/crates/integration-tests/src/test_server.rs +++ b/crates/integration-tests/src/test_server.rs @@ -7,7 +7,7 @@ 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; +use bgp_server::server::BgpServer; use bgp_server::session::SessionRegistry; use bgp_web::{AppState, RouterInfo}; use eyre::Result; @@ -56,7 +56,7 @@ impl TestServer { }; customize(&mut config); - let bound = BgpServerBuilder::new(config).bind().await?; + let bound = BgpServer::bind(config).await?; let addr = bound .local_addrs() .into_iter() diff --git a/crates/integration-tests/tests/announce.rs b/crates/integration-tests/tests/announce.rs index f879dbc..5dcf10f 100644 --- a/crates/integration-tests/tests/announce.rs +++ b/crates/integration-tests/tests/announce.rs @@ -8,7 +8,8 @@ use bgp_integration_tests::test_peer::TestPeer; use bgp_integration_tests::test_server::{RouteChange, TestServer}; use bgp_packet::message::{ AsPathAttribute, CommunitiesPathAttribute, LargeCommunitiesPathAttribute, LargeCommunity, - NextHopPathAttribute, OriginPathAttribute, PathAttribute, UpdateMessage, + LocalPrefPathAttribute, NextHopPathAttribute, OriginPathAttribute, PathAttribute, + UpdateMessage, }; use bgp_primitives::ip_prefix::IpPrefix; use bgp_rib::{LocalRoute, LocalRouteAttrs}; @@ -452,3 +453,48 @@ async fn test_configured_routes_are_announced_to_every_new_session() -> Result<( } Ok(()) } + +/// A peer in our own AS is sent a route we originate with an empty AS path +/// and a LOCAL_PREF, whatever prepending is configured: our AS number goes +/// on the path only as the route leaves the AS. It used to be sent as to +/// an external peer, which an internal one refuses. +#[tokio::test] +async fn test_a_peer_in_our_own_as_gets_an_empty_path_and_a_local_pref() -> Result<()> { + let server = TestServer::start(LOCAL_ASN).await?; + let mut peer = TestPeer::connect(server.addr).await?; + peer.handshake(LOCAL_ASN, Ipv4Addr::new(10, 0, 0, 1), 90) + .await?; + // The peer reads as a router in the server's AS does: one outside it + // would drop the LOCAL_PREF. + peer.ctx.ibgp = true; + wait_for_sessions(&server, 1).await?; + assert!(server.registry.list()[0].ibgp); + + server + .local_rib + .announce(route( + "10.100.1.0/24", + LocalRouteAttrs { + prepend: 2, + ..LocalRouteAttrs::default() + }, + )) + .await?; + let update = next_update(&mut peer).await?; + assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]); + assert!( + update + .path_attributes + .contains(&PathAttribute::AsPath(AsPathAttribute { segments: vec![] })), + "{:?}", + update.path_attributes + ); + assert!( + update + .path_attributes + .contains(&PathAttribute::LocalPref(LocalPrefPathAttribute(100))), + "{:?}", + update.path_attributes + ); + Ok(()) +} diff --git a/crates/integration-tests/tests/dialer.rs b/crates/integration-tests/tests/dialer.rs index 7fd5652..d4e751e 100644 --- a/crates/integration-tests/tests/dialer.rs +++ b/crates/integration-tests/tests/dialer.rs @@ -269,3 +269,51 @@ async fn test_local_address_is_used_as_the_source() -> Result<()> { assert!(reason.contains("local_address 192.0.2.1"), "{reason}"); Ok(()) } + +/// A peer that takes the connection and then refuses the session, here +/// because it is not the AS we are configured to expect, will refuse it +/// again. It is dialled ever more slowly, like a peer that cannot be +/// reached: connecting used to count as success and reset the wait, so +/// such a peer was dialled every few seconds for ever. +#[tokio::test] +async fn test_a_peer_that_refuses_the_session_is_dialled_ever_more_slowly() -> Result<()> { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await?; + let server = server_dialing(&listener, PEER_ASN).await?; + + let mut waits = Vec::new(); + for _ in 0..3 { + let mut peer = accept(&listener).await?; + peer.send_open(65099, Ipv4Addr::new(10, 0, 0, 2), 90) + .await?; + let _ = peer.recv_open().await?; + let refusal = tokio::time::timeout(TIMEOUT, peer.recv()).await??; + assert!( + matches!(&refusal, Message::Notification(n) if n.error == BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs)), + "{refusal:?}" + ); + drop(peer); + + // How long the dialer says it will wait before trying again. + let wait = tokio::time::timeout(TIMEOUT, async { + loop { + if let Some((_, DialState::Backoff { until, last_error })) = + server.registry.configured_peers().into_iter().next() + { + assert!(last_error.contains("AS65099"), "{last_error}"); + return until + .duration_since(std::time::SystemTime::now()) + .unwrap_or_default(); + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await?; + waits.push(wait); + } + // 200 ms, then 400, then 800, less what passed before each was read. + assert!( + waits[2] > Duration::from_millis(500), + "the waits were {waits:?}" + ); + Ok(()) +} diff --git a/crates/integration-tests/tests/interop.rs b/crates/integration-tests/tests/interop.rs index 2e9e53c..d7d410c 100644 --- a/crates/integration-tests/tests/interop.rs +++ b/crates/integration-tests/tests/interop.rs @@ -119,8 +119,16 @@ async fn test_ipv6_prefix_update_shape() -> Result<()> { LocalRouteAttrs::default(), ); let nh: std::net::Ipv6Addr = "fd00:179::2".parse()?; - let update = bgp_server::advertise::build_announce(&route, 4200000001, nh.into(), true) - .expect("ipv6 over ipv6"); + let update = bgp_server::advertise::build_announce( + &route, + &bgp_server::advertise::Towards { + local_asn: 4200000001, + next_hop: nh.into(), + four_octet_asn: true, + ibgp: false, + }, + ) + .expect("ipv6 over ipv6"); assert!(update.announced_routes.is_empty()); assert!( !update diff --git a/crates/packet/src/open.rs b/crates/packet/src/open.rs index 55c0973..b6773f2 100644 --- a/crates/packet/src/open.rs +++ b/crates/packet/src/open.rs @@ -965,10 +965,7 @@ mod tests { assert_eq!(buf.to_vec(), out.to_vec()); Ok(()) } - /// Each capability must leave the bytes that follow it for the caller. - /// Returning the (empty) per-capability buffer instead makes the list - /// parser stop early and silently lose every later capability, e.g. the - /// 4-octet ASN that routers list after graceful restart. + /// A peer advertising Multiple Labels used to hit a `todo!()` and take /// the session task down with it. #[test] @@ -1228,6 +1225,10 @@ mod tests { assert!(too_long.to_wire(&mut BytesMut::new()).is_err()); } + /// Each capability must leave the bytes that follow it for the caller. + /// Returning the (empty) per-capability buffer instead makes the list + /// parser stop early and silently lose every later capability, e.g. the + /// 4-octet ASN that routers list after graceful restart. #[test] fn capability_parsers_leave_the_following_bytes() { let trailer: &[u8] = &[0x41, 0x04, 0x00, 0x03, 0x34, 0x74]; diff --git a/crates/rib/src/lib.rs b/crates/rib/src/lib.rs index 61682b4..dc215a3 100644 --- a/crates/rib/src/lib.rs +++ b/crates/rib/src/lib.rs @@ -3,9 +3,7 @@ mod error; mod local_rib; -mod peer; pub mod table; pub use error::RibStopped; pub use local_rib::{LocalRib, LocalRibEvent, LocalRoute, LocalRouteAttrs}; -pub use peer::PeerId; diff --git a/crates/rib/src/peer.rs b/crates/rib/src/peer.rs deleted file mode 100644 index 808d3aa..0000000 --- a/crates/rib/src/peer.rs +++ /dev/null @@ -1,14 +0,0 @@ -use std::net::SocketAddr; - -/// Identifies a BGP peer by its TCP address and negotiated ASN. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct PeerId { - pub addr: SocketAddr, - pub asn: u32, -} - -impl std::fmt::Display for PeerId { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}(AS{})", self.addr, self.asn) - } -} diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index c606518..6b8b793 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -15,6 +15,7 @@ bytes.workspace = true clap.workspace = true futures.workspace = true prometheus.workspace = true +socket2.workspace = true tokio.workspace = true tracing-subscriber.workspace = true tracing.workspace = true diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 4757086..c17231e 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -11,121 +11,154 @@ use std::net::IpAddr; use bgp_packet::message::{ AsPathAttribute, AsPathSegment, CommunitiesPathAttribute, LargeCommunitiesPathAttribute, - Message, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, MultiExitDiscPathAttribute, - NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage, + LocalPrefPathAttribute, Message, MpReachNlriPathAttribute, MpUnreachNlriPathAttribute, + MultiExitDiscPathAttribute, NextHopPathAttribute, NlriNextHop, PathAttribute, UpdateMessage, }; use bgp_primitives::constants::{AfiSafi, IpVersion}; use bgp_primitives::ip_prefix::{IpPrefix, Nlri}; +use bgp_rib::table::DEFAULT_LOCAL_PREF; use bgp_rib::{LocalRib, LocalRibEvent, LocalRoute, RibStopped}; +use tokio::sync::broadcast::Receiver; use tokio::sync::broadcast::error::RecvError; use tokio::task::JoinHandle; use tracing::{debug, info, warn}; use crate::session::{SessionEvent, SessionHandle, SessionRegistry}; +/// Starts the advertiser. It hears of every session that comes up after +/// this returns, so call it before anything that can bring one up. pub fn spawn_advertiser( local_rib: LocalRib, registry: SessionRegistry, local_asn: u32, ) -> JoinHandle<()> { - tokio::spawn(run(local_rib, registry, local_asn)) -} - -/// Runs until the local RIB stops. The server watches for this task ending: -/// without it nothing is announced. -async fn run(local_rib: LocalRib, registry: SessionRegistry, local_asn: u32) { - let mut rib_events = local_rib.subscribe(); - let mut session_events = registry.subscribe(); - - loop { - tokio::select! { - ev = rib_events.recv() => match ev { - Ok(LocalRibEvent::Announced(route)) => { - for session in registry.list() { - send_route(&session, &route, local_asn).await; - } - } - Ok(LocalRibEvent::Withdrawn { prefix }) => { - for session in registry.list() { - send_withdraw(&session, &prefix).await; - } - } - Err(RecvError::Lagged(n)) => { - warn!(missed = n, "Advertiser lagged on local RIB events, resyncing all sessions"); - if resync_all(&local_rib, ®istry, local_asn).await.is_err() { - return; - } - } - Err(RecvError::Closed) => return, - }, - ev = session_events.recv() => match ev { - Ok(SessionEvent::Up(session)) => { - let Ok(routes) = local_rib.snapshot().await else { return }; - info!(peer = %session.peer, routes = routes.len(), "Session up, sending local routes"); - for route in &routes { - send_route(&session, route, local_asn).await; - } - } - Ok(SessionEvent::RefreshRequested(session)) => { - let Ok(routes) = local_rib.snapshot().await else { return }; - info!(peer = %session.peer, routes = routes.len(), "Route refresh requested, re-sending local routes"); - for route in &routes { - send_route(&session, route, local_asn).await; - } - } - Ok(SessionEvent::Down(_)) => {} - Err(RecvError::Lagged(n)) => { - warn!(missed = n, "Advertiser lagged on session events, resyncing all sessions"); - if resync_all(&local_rib, ®istry, local_asn).await.is_err() { - return; - } - } - Err(RecvError::Closed) => return, - }, - } - } -} - -async fn resync_all( - local_rib: &LocalRib, - registry: &SessionRegistry, - local_asn: u32, -) -> Result<(), RibStopped> { - let routes = local_rib.snapshot().await?; - for session in registry.list() { - for route in &routes { - send_route(&session, route, local_asn).await; - } - } - Ok(()) -} - -async fn send_route(session: &SessionHandle, route: &LocalRoute, local_asn: u32) { - if !session.families.carries(&route.prefix) { - debug!( - peer = %session.peer, - prefix = %route.prefix, - "Skipping route: its address family was not negotiated" - ); - return; - } - match build_announce( - route, + let advertiser = Advertiser { + rib_events: local_rib.subscribe(), + session_events: registry.subscribe(), + local_rib, + registry, local_asn, - session.local_addr.ip(), - session.four_octet_asn, - ) { - Some(update) => { - debug!(peer = %session.peer, prefix = %route.prefix, "Announcing local route"); - if let Err(e) = session.send(Message::Update(update)).await { - warn!(peer = %session.peer, err = %e, "Failed to queue announce"); + }; + tokio::spawn(advertiser.run()) +} + +struct Advertiser { + local_rib: LocalRib, + registry: SessionRegistry, + local_asn: u32, + rib_events: Receiver, + session_events: Receiver, +} + +impl Advertiser { + /// Runs until the local RIB stops. The server watches for this task + /// ending: without it nothing is announced. + async fn run(mut self) { + loop { + let handled = tokio::select! { + event = self.rib_events.recv() => self.on_route_event(event).await, + event = self.session_events.recv() => self.on_session_event(event).await, + }; + if handled.is_err() { + return; } } - None => debug!( - peer = %session.peer, - prefix = %route.prefix, - "Skipping route: address family does not match session transport" - ), + } + + async fn on_route_event( + &self, + event: Result, + ) -> Result<(), RibStopped> { + match event { + Ok(LocalRibEvent::Announced(route)) => { + for session in self.registry.list() { + self.send_route(&session, &route).await; + } + Ok(()) + } + Ok(LocalRibEvent::Withdrawn { prefix }) => { + for session in self.registry.list() { + send_withdraw(&session, &prefix).await; + } + Ok(()) + } + Err(RecvError::Lagged(missed)) => { + warn!( + missed, + "Advertiser lagged on local RIB events, resyncing all sessions" + ); + self.send_all_to(&self.registry.list()).await + } + Err(RecvError::Closed) => Err(RibStopped { rib: "local RIB" }), + } + } + + async fn on_session_event( + &self, + event: Result, + ) -> Result<(), RibStopped> { + match event { + Ok(SessionEvent::Up(session)) => { + info!(peer = %session.peer, "Session up, sending local routes"); + self.send_all_to(&[session]).await + } + Ok(SessionEvent::RefreshRequested(session)) => { + info!(peer = %session.peer, "Route refresh requested, re-sending local routes"); + self.send_all_to(&[session]).await + } + Ok(SessionEvent::Down(_)) => Ok(()), + Err(RecvError::Lagged(missed)) => { + warn!( + missed, + "Advertiser lagged on session events, resyncing all sessions" + ); + self.send_all_to(&self.registry.list()).await + } + // The registry outlives the server's tasks; nothing more will + // come up, and nothing more needs sending. + Err(RecvError::Closed) => Err(RibStopped { rib: "local RIB" }), + } + } + + /// Sends every originated route to each of `sessions`. + async fn send_all_to(&self, sessions: &[SessionHandle]) -> Result<(), RibStopped> { + let routes = self.local_rib.snapshot().await?; + for session in sessions { + for route in &routes { + self.send_route(session, route).await; + } + } + Ok(()) + } + + async fn send_route(&self, session: &SessionHandle, route: &LocalRoute) { + if !session.families.carries(&route.prefix) { + debug!( + peer = %session.peer, + prefix = %route.prefix, + "Skipping route: its address family was not negotiated" + ); + return; + } + let towards = Towards { + local_asn: self.local_asn, + next_hop: session.local_addr.ip(), + four_octet_asn: session.four_octet_asn, + ibgp: session.ibgp, + }; + match build_announce(route, &towards) { + Some(update) => { + debug!(peer = %session.peer, prefix = %route.prefix, "Announcing local route"); + if let Err(e) = session.send(Message::Update(update)).await { + warn!(peer = %session.peer, err = %e, "Failed to queue announce"); + } + } + None => debug!( + peer = %session.peer, + prefix = %route.prefix, + "Skipping route: address family does not match session transport" + ), + } } } @@ -140,31 +173,44 @@ async fn send_withdraw(session: &SessionHandle, prefix: &IpPrefix) { } } -/// Builds the UPDATE announcing `route` with `next_hop` as the next hop. +/// What about a session decides how a route is announced on it. +#[derive(Debug, Clone, Copy)] +pub struct Towards { + pub local_asn: u32, + /// Our address on the session, which is the routes' next hop. + pub next_hop: IpAddr, + /// Whether the session negotiated four-octet AS numbers. If it did + /// not, and an AS number in the path needs four octets, the path is + /// also sent as AS4_PATH so that it survives the two-octet speaker + /// (RFC 6793 §4.2.2). + pub four_octet_asn: bool, + /// Whether the peer is in our own AS. + pub ibgp: bool, +} + +/// Builds the UPDATE announcing `route` on a session. /// /// Returns `None` when the prefix family does not match the next-hop family: /// carrying IPv4 NLRI over an IPv6 next hop (RFC 8950) or vice versa is not /// supported yet. -/// -/// `four_octet_asn` is whether the session the UPDATE is for negotiated -/// four-octet AS numbers. If it did not, and the local AS number needs four -/// octets, the path is also sent as AS4_PATH so that it survives the -/// two-octet speaker (RFC 6793 §4.2.2). -pub fn build_announce( - route: &LocalRoute, - local_asn: u32, - next_hop: IpAddr, - four_octet_asn: bool, -) -> Option { +pub fn build_announce(route: &LocalRoute, towards: &Towards) -> Option { let attrs = &route.attrs; + // Our AS number goes on the path as a route leaves the AS, and not + // before (RFC 4271 §5.1.2): a route we originate has an empty path + // inside it. A peer in our AS that found our AS number on the path + // would take the route for one that had looped. let as_path = AsPathAttribute { - segments: vec![AsPathSegment { - ordered: true, - path: vec![local_asn; 1 + usize::from(attrs.prepend)], - }], + segments: if towards.ibgp { + vec![] + } else { + vec![AsPathSegment { + ordered: true, + path: vec![towards.local_asn; 1 + usize::from(attrs.prepend)], + }] + }, }; - let as4_path = if four_octet_asn { + let as4_path = if towards.four_octet_asn { None } else { as_path.as4_path() @@ -176,34 +222,42 @@ pub fn build_announce( PathAttribute::Origin(attrs.origin), PathAttribute::AsPath(as_path), ]; - + // IPv4 routes go in the body of the UPDATE with their next hop in + // NEXT_HOP; IPv6 routes go in MP_REACH_NLRI, which carries its own. let mut announced_routes = vec![]; - match (route.prefix.address_family(), next_hop) { - (IpVersion::Ipv4, IpAddr::V4(nh)) => { - path_attributes.push(PathAttribute::NextHop(NextHopPathAttribute(nh))); + let mp_reach = match (route.prefix.address_family(), towards.next_hop) { + (IpVersion::Ipv4, IpAddr::V4(next_hop)) => { + path_attributes.push(PathAttribute::NextHop(NextHopPathAttribute(next_hop))); announced_routes.push(route.prefix.into()); + None + } + (IpVersion::Ipv6, IpAddr::V6(next_hop)) => { + Some(PathAttribute::MpReachNlri(MpReachNlriPathAttribute { + family: AfiSafi::IPV6_UNICAST, + next_hop: NlriNextHop::Ipv6(next_hop), + prefixes: vec![route.prefix.into()], + })) } - (IpVersion::Ipv6, IpAddr::V6(_)) => {} _ => return None, - } - + }; if let Some(med) = attrs.med { path_attributes.push(PathAttribute::MultiExitDisc(MultiExitDiscPathAttribute( med, ))); } + // LOCAL_PREF is sent to peers in our AS and to no others (RFC 4271 + // §5.1.5). + if towards.ibgp { + path_attributes.push(PathAttribute::LocalPref(LocalPrefPathAttribute( + DEFAULT_LOCAL_PREF, + ))); + } if !attrs.communities.is_empty() { path_attributes.push(PathAttribute::Communities(CommunitiesPathAttribute( attrs.communities.clone(), ))); } - if let (IpVersion::Ipv6, IpAddr::V6(nh)) = (route.prefix.address_family(), next_hop) { - path_attributes.push(PathAttribute::MpReachNlri(MpReachNlriPathAttribute { - family: AfiSafi::IPV6_UNICAST, - next_hop: NlriNextHop::Ipv6(nh), - prefixes: vec![route.prefix.into()], - })); - } + path_attributes.extend(mp_reach); path_attributes.extend(as4_path); if !attrs.large_communities.is_empty() { path_attributes.push(PathAttribute::LargeCommunities( @@ -254,6 +308,16 @@ mod tests { IpPrefix::try_from(s).unwrap() } + /// An external peer. + fn towards(local_asn: u32, next_hop: IpAddr, four_octet_asn: bool) -> Towards { + Towards { + local_asn, + next_hop, + four_octet_asn, + ibgp: false, + } + } + #[test] fn ipv4_announce_has_expected_attributes_and_roundtrips() { let route = LocalRoute::new( @@ -270,8 +334,11 @@ mod tests { }], }, ); - let update = - build_announce(&route, 65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true).unwrap(); + let update = build_announce( + &route, + &towards(65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true), + ) + .unwrap(); assert_eq!(update.announced_routes, vec![prefix("10.100.1.0/24")]); assert!(update.withdrawn_routes.is_empty()); @@ -315,7 +382,7 @@ mod tests { fn ipv6_announce_uses_mp_reach() { let route = LocalRoute::new(prefix("2001:db8:1::/48"), LocalRouteAttrs::default()); let nh = Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1); - let update = build_announce(&route, 65000, IpAddr::V6(nh), true).unwrap(); + let update = build_announce(&route, &towards(65000, IpAddr::V6(nh), true)).unwrap(); assert!(update.announced_routes.is_empty()); assert!( @@ -351,7 +418,7 @@ mod tests { }) }; - let update = build_announce(&route, 4_200_000_001, next_hop, false).unwrap(); + let update = build_announce(&route, &towards(4_200_000_001, next_hop, false)).unwrap(); assert_eq!(path(&update, true), Some(vec![4_200_000_001; 2])); // What the peer reads from the wire. @@ -370,9 +437,9 @@ mod tests { // No AS4_PATH when it would say nothing new: the peer handles // four-octet numbers, or ours fits in two. - let update = build_announce(&route, 4_200_000_001, next_hop, true).unwrap(); + let update = build_announce(&route, &towards(4_200_000_001, next_hop, true)).unwrap(); assert_eq!(path(&update, true), None); - let update = build_announce(&route, 65000, next_hop, false).unwrap(); + let update = build_announce(&route, &towards(65000, next_hop, false)).unwrap(); assert_eq!(path(&update, true), None); assert_eq!(path(&update, false), Some(vec![65000; 2])); } @@ -380,9 +447,13 @@ mod tests { #[test] fn family_mismatch_is_skipped() { let v6 = LocalRoute::new(prefix("2001:db8::/32"), LocalRouteAttrs::default()); - assert!(build_announce(&v6, 65000, IpAddr::V4(Ipv4Addr::LOCALHOST), true).is_none()); + assert!( + build_announce(&v6, &towards(65000, IpAddr::V4(Ipv4Addr::LOCALHOST), true)).is_none() + ); let v4 = LocalRoute::new(prefix("10.0.0.0/8"), LocalRouteAttrs::default()); - assert!(build_announce(&v4, 65000, IpAddr::V6(Ipv6Addr::LOCALHOST), true).is_none()); + assert!( + build_announce(&v4, &towards(65000, IpAddr::V6(Ipv6Addr::LOCALHOST), true)).is_none() + ); } #[test] @@ -397,4 +468,69 @@ mod tests { matches!(&v6.path_attributes[..], [PathAttribute::MpUnreachNlri(m)] if m.prefixes == vec![prefix("2001:db8::/32")]) ); } + + /// Inside our own AS a route we originate has an empty AS path and a + /// LOCAL_PREF. With our AS number on the path, a peer in our AS takes + /// the route for a loop; without LOCAL_PREF it treats the route as + /// withdrawn, as our own decoder does. + #[test] + fn a_route_announced_inside_our_as_has_an_empty_path_and_a_local_pref() { + let route = LocalRoute::new( + prefix("10.100.1.0/24"), + LocalRouteAttrs { + prepend: 3, + ..LocalRouteAttrs::default() + }, + ); + let internal = Towards { + ibgp: true, + ..towards(65000, IpAddr::V4(Ipv4Addr::new(192, 0, 2, 1)), true) + }; + let update = build_announce(&route, &internal).unwrap(); + assert_eq!( + update.path_attributes, + vec![ + PathAttribute::Origin(OriginPathAttribute::IGP), + PathAttribute::AsPath(AsPathAttribute { segments: vec![] }), + PathAttribute::NextHop(NextHopPathAttribute(Ipv4Addr::new(192, 0, 2, 1))), + PathAttribute::LocalPref(LocalPrefPathAttribute(100)), + ] + ); + + // An internal peer accepts it as it stands. + let ctx = ParserContext { + four_octet_asn: true, + ibgp: true, + add_path: Vec::new(), + }; + let mut buf = BytesMut::new(); + Message::Update(update.clone()) + .to_wire(&ctx, &mut buf) + .unwrap(); + let decoded = Message::decode_with_report(&ctx, &buf).unwrap(); + assert!( + decoded.update_errors.is_empty(), + "{:?}", + decoded.update_errors + ); + assert_eq!(decoded.message, Message::Update(update)); + + // To an external peer the same route has our AS on the path, as + // many times as configured, and no LOCAL_PREF. + let external = Towards { + ibgp: false, + ..internal + }; + let update = build_announce(&route, &external).unwrap(); + assert_eq!( + update.path_attributes[1], + AsPathAttribute::from_asns(vec![65000; 4]) + ); + assert!( + !update + .path_attributes + .iter() + .any(|a| matches!(a, PathAttribute::LocalPref(_))) + ); + } } diff --git a/crates/server/src/collision.rs b/crates/server/src/collision.rs index 4dd50f7..f067391 100644 --- a/crates/server/src/collision.rs +++ b/crates/server/src/collision.rs @@ -8,9 +8,14 @@ //! This cannot be a check local to one connection: the rival connection is //! usually still mid-handshake (so it is not in the session registry yet), //! and sometimes it is the rival, running in another task, that must close. -//! Hence one small shared map, keyed by the peer's BGP Identifier, holding -//! the single connection currently preferred for that peer and a way to -//! tell it to stop. +//! Hence one small shared map, holding the single connection currently +//! preferred for each peer and a way to tell it to stop. +//! +//! A peer here is an address together with the BGP Identifier it announced. +//! The identifier alone would not do: a router commonly has one session +//! over IPv4 and another over IPv6 with the same neighbour, which are two +//! sessions and not a collision; and it would let any host that can reach +//! us displace a peer's connection by claiming the peer's identifier. //! //! A collision with an *Established* connection closes the new one, unless //! `detect_established` is set (the RFC's optional @@ -19,7 +24,7 @@ //! by the same side the newer wins: the initiator has abandoned the older. use std::collections::HashMap; -use std::net::Ipv4Addr; +use std::net::{IpAddr, Ipv4Addr}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; @@ -76,10 +81,14 @@ fn new_loses( new != keep } +/// Who a connection is with: the address at the other end, and the BGP +/// Identifier announced from it. +type Peer = (IpAddr, Ipv4Addr); + /// Shared by every connection of one server. #[derive(Clone)] pub struct CollisionTable { - entries: Arc>>, + entries: Arc>>, next_id: Arc, detect_established: bool, } @@ -100,11 +109,13 @@ impl CollisionTable { pub fn admit( &self, local_id: Ipv4Addr, + remote_addr: IpAddr, remote_id: Ipv4Addr, direction: Direction, ) -> Option { + let peer = (remote_addr, remote_id); let mut entries = self.entries.locked(); - if let Some(existing) = entries.get_mut(&remote_id) { + if let Some(existing) = entries.get_mut(&peer) { if new_loses( local_id, remote_id, @@ -123,7 +134,7 @@ impl CollisionTable { let id = self.next_id.fetch_add(1, Ordering::Relaxed); let (kill, killed) = oneshot::channel(); entries.insert( - remote_id, + peer, Entry { id, direction, @@ -133,7 +144,7 @@ impl CollisionTable { ); Some(ConnectionGuard { table: self.clone(), - remote_id, + peer, id, killed, }) @@ -143,20 +154,28 @@ impl CollisionTable { /// Holds a connection's slot in the table; releases it on drop. pub struct ConnectionGuard { table: CollisionTable, - remote_id: Ipv4Addr, + peer: Peer, id: u64, - /// Completes when a later connection won against this one; the owner - /// must then send Cease / Connection Collision Resolution and close. - pub killed: oneshot::Receiver<()>, + killed: oneshot::Receiver<()>, } impl ConnectionGuard { pub fn set_established(&self) { let mut entries = self.table.entries.locked(); - if let Some(e) = entries.get_mut(&self.remote_id).filter(|e| e.id == self.id) { + if let Some(e) = entries.get_mut(&self.peer).filter(|e| e.id == self.id) { e.established = true; } } + + /// Completes when a later connection has won against this one; the + /// owner must then send Cease / Connection Collision Resolution and + /// close. Cancel-safe. + pub async fn killed(&mut self) { + // The sending half is dropped only when it has been used: the entry + // that holds it is removed by this guard alone, or replaced by a + // connection that signals first. Either way, the wait is over. + let _ = (&mut self.killed).await; + } } impl Drop for ConnectionGuard { @@ -164,11 +183,8 @@ impl Drop for ConnectionGuard { // A connection that lost has already been replaced in the map; only // remove the slot if it is still ours. let mut entries = self.table.entries.locked(); - if entries - .get(&self.remote_id) - .is_some_and(|e| e.id == self.id) - { - entries.remove(&self.remote_id); + if entries.get(&self.peer).is_some_and(|e| e.id == self.id) { + entries.remove(&self.peer); } } } @@ -247,24 +263,42 @@ mod tests { #[tokio::test] async fn table_signals_loser_and_releases_slots() { let table = CollisionTable::new(false); + let addr: IpAddr = Ipv4Addr::new(192, 0, 2, 1).into(); // We are LOW: outbound first, then the peer's inbound wins. - let mut out = table.admit(LOW, HIGH, Outbound).unwrap(); - let inb = table.admit(LOW, HIGH, Inbound).unwrap(); + let mut out = table.admit(LOW, addr, HIGH, Outbound).unwrap(); + let inb = table.admit(LOW, addr, HIGH, Inbound).unwrap(); assert!(out.killed.try_recv().is_ok()); // The loser's guard dropping must not free the winner's slot. drop(out); inb.set_established(); - assert!(table.admit(LOW, HIGH, Outbound).is_none()); + assert!(table.admit(LOW, addr, HIGH, Outbound).is_none()); // A different identifier never collides. assert!( table - .admit(LOW, Ipv4Addr::new(192, 0, 2, 2), Inbound) + .admit(LOW, addr, Ipv4Addr::new(192, 0, 2, 2), Inbound) .is_some() ); // Once the survivor goes away the identifier is free again. drop(inb); - assert!(table.admit(LOW, HIGH, Outbound).is_some()); + assert!(table.admit(LOW, addr, HIGH, Outbound).is_some()); + } + + /// A neighbour reached over IPv4 and over IPv6 announces the same + /// identifier on both sessions. They are two sessions, and neither + /// closes the other. Nor can a connection from elsewhere that claims a + /// peer's identifier take the place of the peer's own. + #[tokio::test] + async fn the_same_identifier_from_another_address_is_another_peer() { + let table = CollisionTable::new(true); + let v4: IpAddr = Ipv4Addr::new(192, 0, 2, 1).into(); + let v6: IpAddr = "2001:db8::1".parse().unwrap(); + + let mut over_v4 = table.admit(LOW, v4, HIGH, Outbound).unwrap(); + over_v4.set_established(); + let over_v6 = table.admit(LOW, v6, HIGH, Inbound); + assert!(over_v6.is_some()); + assert!(over_v4.killed.try_recv().is_err(), "the first is untouched"); } } diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index b22ebfb2..d7325a9 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -56,9 +56,17 @@ pub struct PeerConfig { } impl Config { + /// The configured peer at `address`, if there is one. + pub fn peer_at(&self, address: IpAddr) -> Option<&PeerConfig> { + self.peers.iter().find(|peer| peer.endpoint.ip() == address) + } + pub fn from_proto(proto: BgpConfig) -> Result { + // Peers refuse an identifier of 0.0.0.0, as we do theirs. let router_id = Ipv4Addr::from_str(&proto.router_id) - .map_err(|_| ConfigError::RouterId(proto.router_id.clone()))?; + .ok() + .filter(|id| !id.is_unspecified()) + .ok_or_else(|| ConfigError::RouterId(proto.router_id.clone()))?; if proto.local_asn == 0 { return Err(ConfigError::LocalAsn); @@ -80,17 +88,27 @@ impl Config { .collect::>()? }; - let peers = proto - .peers - .into_iter() - .map(|peer| { - let name = peer.name.clone(); - parse_peer(peer).map_err(|problem| ConfigError::Peer { - peer: name, - problem, - }) - }) - .collect::>()?; + let mut peers: Vec = Vec::with_capacity(proto.peers.len()); + for peer in proto.peers { + let name = peer.name.clone(); + let rejected = |problem| ConfigError::Peer { + peer: name.clone(), + problem, + }; + let peer = parse_peer(peer).map_err(rejected)?; + // A peer is told from the others by its name in what the + // dialer reports, and by its address when it connects to us. + if peers.iter().any(|earlier| earlier.name == peer.name) { + return Err(rejected(PeerProblem::DuplicateName)); + } + if peers + .iter() + .any(|earlier| earlier.endpoint.ip() == peer.endpoint.ip()) + { + return Err(rejected(PeerProblem::DuplicateAddress)); + } + peers.push(peer); + } let mut originate: Vec = Vec::with_capacity(proto.originate.len()); for entry in &proto.originate { @@ -185,7 +203,12 @@ fn parse_peer(peer: Peer) -> Result { }; Ok(PeerConfig { - name: peer.name, + // A peer without a name goes by its address. + name: if peer.name.is_empty() { + endpoint.ip().to_string() + } else { + peer.name + }, endpoint, remote_asn: peer.remote_asn, hold_time_secs, @@ -421,4 +444,54 @@ mod tests { ConfigError::HttpPort ); } + + /// Peers are told apart by name where the dialer reports on them and by + /// address when they connect to us; two that share either would be + /// taken for each other. + #[test] + fn peers_must_differ_in_name_and_address() { + let peer = |name: &str, address: &str| { + format!( + "peers {{ name: \"{name}\" endpoint {{ address: \"{address}\" }} remote_asn: 1 }}\n" + ) + }; + let problem = |peers: String| match rejected(&format!("{BASE}{peers}")) { + ConfigError::Peer { problem, .. } => problem, + other => panic!("{other}"), + }; + assert_eq!( + problem(peer("r1", "192.0.2.2") + &peer("r1", "192.0.2.3")), + PeerProblem::DuplicateName + ); + assert_eq!( + problem(peer("r1", "192.0.2.2") + &peer("r2", "192.0.2.2")), + PeerProblem::DuplicateAddress + ); + + // A peer without a name goes by its address, so two of them differ. + let config = parse(&format!( + "{BASE}{}{}", + peer("", "192.0.2.2"), + peer("", "192.0.2.3") + )) + .unwrap(); + assert_eq!(config.peers[0].name, "192.0.2.2"); + assert_eq!(config.peers[1].name, "192.0.2.3"); + assert_eq!( + config + .peer_at("192.0.2.3".parse().unwrap()) + .map(|p| &*p.name), + Some("192.0.2.3") + ); + assert!(config.peer_at("192.0.2.4".parse().unwrap()).is_none()); + } + + /// A peer would refuse our OPEN. + #[test] + fn a_router_id_of_zero_is_rejected() { + assert_eq!( + rejected("router_id: \"0.0.0.0\" local_asn: 1"), + ConfigError::RouterId("0.0.0.0".into()) + ); + } } diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index ff927d5..78f1c96 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -11,6 +11,7 @@ use std::io; use std::net::{Ipv4Addr, SocketAddr}; use std::pin::Pin; use std::sync::Arc; +use std::time::Duration; use bgp_packet::errors::{BgpError, DecodeError, MsgHeaderSubcode}; use bgp_packet::message::{ @@ -19,78 +20,129 @@ use bgp_packet::message::{ use bgp_packet::open::AddPathSendRecv; use bgp_packet::update_decode::UpdateError; use bgp_primitives::parser::ParserContext; -use bgp_rib::PeerId; -use bgp_rib::table::{Outcome, PeerIndex, PeerInfo, RibTable, TooManyPeers}; +use bgp_rib::table::{Outcome, PeerIndex, PeerInfo, RibTable}; 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 tokio::time::{Instant, Interval, MissedTickBehavior, Sleep, interval_at, sleep, timeout}; use tracing::{debug, error, info, warn}; use crate::collision::{CollisionTable, ConnectionGuard, Direction}; +use crate::config::PeerConfig; use crate::error::SessionError; use crate::family::FamilySet; -use crate::fsm::{Action, Event, Fsm, FsmConfig, Negotiated}; +use crate::fsm::{Action, Event, Fsm, FsmConfig, Negotiated, Received}; 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}; +use crate::session::{PeerId, SessionGuard, SessionHandle, SessionRegistry}; /// Number of outbound messages that may be queued for a session before /// senders are back-pressured. const OUTBOUND_QUEUE_CAPACITY: usize = 1024; -/// What a session needs from the rest of the server. -#[derive(Clone)] -pub struct SessionConfig { +/// The least time a write to the peer may make no progress before the +/// session is given up: the Send Hold Timer of RFC 9687, which is this or +/// twice the hold time, whichever is longer. A peer that has stopped +/// reading would otherwise hold the session, and our routes, for as long +/// as TCP keeps trying. +const MIN_SEND_HOLD_TIME: Duration = Duration::from_secs(8 * 60); + +/// How long a connection we are closing is kept to let the peer read our +/// NOTIFICATION. Closing at once, with something of the peer's still +/// unread, resets the connection, and the peer may then never see why. +const CLOSE_LINGER: Duration = Duration::from_secs(1); + +/// What every session of a server shares. +pub struct ServerContext { pub router_id: Ipv4Addr, pub local_asn: u32, - pub hold_time_secs: u16, - /// Where routes learned from the peer go. + /// Where routes learned from peers go. pub rib: Arc, pub registry: SessionRegistry, - /// Shared table used to resolve connection collisions (RFC 4271 §6.8). + /// Used to resolve connection collisions (RFC 4271 §6.8). pub collisions: CollisionTable, - /// Which side opened this TCP connection. - pub direction: Direction, - /// Configured name of the peer, if known. - pub peer_name: Option, +} + +/// What is configured for the peer at the other end of a connection. +#[derive(Debug, Clone)] +pub struct PeerSettings { + /// Configured name of the peer, if it is a configured one. + pub name: Option, /// If set, the peer's OPEN must carry this ASN or the session is /// refused with a Bad Peer AS notification. pub expected_asn: Option, + pub hold_time_secs: u16, /// The address families to offer the peer. pub families: FamilySet, /// The directions to offer the peer ADD-PATH for (RFC 7911). pub add_path: Option, } -/// Runs a session on an accepted connection and logs how it ended. -pub async fn handle_connection( - stream: TcpStream, - peer_addr: SocketAddr, - config: Arc, -) { - if let Err(e) = run_session(stream, peer_addr, (*config).clone()).await { - warn!(%peer_addr, err = %e, "BGP session terminated with error"); +impl PeerSettings { + /// For a peer that connected to us and is not in the configuration. + pub fn unconfigured(hold_time_secs: u16) -> Self { + Self { + name: None, + expected_asn: None, + hold_time_secs, + families: FamilySet::supported(), + add_path: None, + } } } -/// Runs the BGP session on `stream` until it ends. +impl From<&PeerConfig> for PeerSettings { + fn from(peer: &PeerConfig) -> Self { + Self { + name: Some(peer.name.clone()), + expected_asn: Some(peer.remote_asn), + hold_time_secs: peer.hold_time_secs, + families: peer.families.clone(), + add_path: peer.add_path, + } + } +} + +/// How a session ended. +#[derive(Debug)] +pub struct SessionEnd { + /// Whether it had reached Established. + pub established: bool, + /// `Ok` if it was closed at our own request (a shutdown); otherwise + /// what went wrong, or what the peer said. + pub result: Result<(), SessionError>, +} + +/// Runs the BGP session on `stream` until it ends. `direction` is which +/// side opened the connection. pub async fn run_session( stream: TcpStream, peer_addr: SocketAddr, - config: SessionConfig, -) -> Result<(), SessionError> { - let (mut session, first_actions) = Session::open(stream, peer_addr, config)?; + direction: Direction, + server: Arc, + peer: PeerSettings, +) -> SessionEnd { + let (mut session, first_actions) = + match Session::open(stream, peer_addr, direction, server, peer) { + Ok(opened) => opened, + Err(e) => { + return SessionEnd { + established: false, + result: Err(e.into()), + }; + } + }; let result = match contained(session.run(first_actions)).await { Ok(result) => result, Err(panic) => Err(session.close_after_panic(panic).await), }; + let established = session.established.is_some(); // Whatever ended the session, routes learned over it are gone // (RFC 4271 §8.2.2). Do this before the registration drops so that the @@ -110,43 +162,10 @@ pub async fn run_session( "Session ended" ); } - result -} - -impl Session { - /// A session on a connection that has just been opened, and the first - /// things to do on it. - fn open( - stream: TcpStream, - peer_addr: SocketAddr, - config: SessionConfig, - ) -> io::Result<(Self, Vec)> { - let local_addr = stream.local_addr()?; - let (reader, writer) = stream.into_split(); - let (fsm, first_actions) = Fsm::new(FsmConfig { - router_id: config.router_id, - local_asn: config.local_asn, - hold_time_secs: config.hold_time_secs, - expected_asn: config.expected_asn, - families: config.families.clone(), - add_path: config.add_path, - }); - let session = Session { - fsm, - frames: FrameReader::new(reader), - writer, - ctx: ParserContext::default(), - send_ctx: ParserContext::default(), - hold_timer: None, - keepalive_timer: None, - collision: None, - established: None, - attributes_wire: None, - peer_addr, - local_addr, - config, - }; - Ok((session, first_actions)) + session.close().await; + SessionEnd { + established, + result, } } @@ -231,10 +250,12 @@ impl Drop for PeerRoutes { } } -struct Session { +struct Session { fsm: Fsm, - frames: FrameReader, + frames: FrameReader, writer: OwnedWriteHalf, + /// Reused for every message written. + write_buf: BytesMut, /// How messages from the peer are decoded: the width of AS numbers and /// where prefixes carry path identifiers are known once the OPENs are /// exchanged. @@ -245,19 +266,63 @@ struct Session { send_ctx: ParserContext, hold_timer: Option>>, keepalive_timer: Option, + /// How long a write may make no progress; see [`MIN_SEND_HOLD_TIME`]. + send_hold_time: Duration, + /// Whether a NOTIFICATION of ours is on its way to the peer. + notified: bool, /// This connection's slot in the collision table, once admitted. collision: Option, established: Option, - /// The attributes of the UPDATE that is being acted on, as they were - /// on the wire, if the decoder kept them. Set when the message is read - /// and taken when its routes are stored. - attributes_wire: Option>, peer_addr: SocketAddr, local_addr: SocketAddr, - config: SessionConfig, + direction: Direction, + server: Arc, + /// Configured name of the peer, if it is a configured one. + peer_name: Option, } -impl Session { +impl Session { + /// A session on a connection that has just been opened, and the first + /// things to do on it. + fn open( + stream: TcpStream, + peer_addr: SocketAddr, + direction: Direction, + server: Arc, + peer: PeerSettings, + ) -> io::Result<(Self, Vec)> { + let local_addr = stream.local_addr()?; + let (reader, writer) = stream.into_split(); + let (fsm, first_actions) = Fsm::new(FsmConfig { + router_id: server.router_id, + local_asn: server.local_asn, + hold_time_secs: peer.hold_time_secs, + expected_asn: peer.expected_asn, + families: peer.families, + add_path: peer.add_path, + }); + let session = Session { + fsm, + frames: FrameReader::new(reader), + writer, + write_buf: BytesMut::new(), + ctx: ParserContext::default(), + send_ctx: ParserContext::default(), + hold_timer: None, + keepalive_timer: None, + send_hold_time: MIN_SEND_HOLD_TIME, + notified: false, + collision: None, + established: None, + peer_addr, + local_addr, + direction, + server, + peer_name: peer.name, + }; + Ok((session, first_actions)) + } + /// Says goodbye to the peer after the session's own code panicked. /// /// The state machine may be half way through something, so it is not @@ -277,6 +342,20 @@ impl Session { error } + /// Closes the connection. If the peer was sent a NOTIFICATION, it is + /// given a moment to read it. + async fn close(mut self) { + // A new connection from the peer must not find this one in its way. + self.collision = None; + if !self.notified { + return; + } + // Our side is done sending; what the peer still sends is read and + // thrown away until it closes its side too. + let _ = self.writer.shutdown().await; + let _ = timeout(CLOSE_LINGER, self.frames.discard_until_closed()).await; + } + async fn run(&mut self, first_actions: Vec) -> Result<(), SessionError> { let mut actions = first_actions; loop { @@ -295,8 +374,10 @@ impl Session { frame = self.frames.next(&self.ctx) => match frame { Ok(decoded) => { self.report(&decoded.update_errors); - self.attributes_wire = decoded.attributes_wire; - Ok(Event::Message(decoded.message)) + Ok(Event::Message(Received { + message: decoded.message, + attributes_wire: decoded.attributes_wire, + })) } Err(SessionError::Decode(e)) => Ok(Event::DecodeFailed(e)), Err(e) => Err(e), @@ -332,17 +413,26 @@ impl Session { let mut queue: VecDeque = actions.into(); while let Some(action) = queue.pop_front() { match action { - Action::Send(msg) => { - let closing = matches!(msg, Message::Notification(_)); - if let Err(e) = self.send(&msg).await { - if !closing { - return Some(Err(e)); - } - // The session is ending anyway and the Close that - // follows says why; a NOTIFICATION is best effort. + Action::Send(msg) => match self.send(&msg).await { + Ok(()) => {} + // The session is ending anyway and the Close that + // follows says why; a NOTIFICATION is best effort. + Err(e) if matches!(msg, Message::Notification(_)) => { warn!(peer_addr = %self.peer_addr, err = %e, "Failed to send NOTIFICATION"); } - } + // A message of our own making that cannot be written. + // That is a bug here, and nothing the peer did: the + // session is worth more than the message. + Err(SessionError::Encode(e)) => { + error!( + peer_addr = %self.peer_addr, + message = message_type_name(&msg), + err = %e, + "Dropped a message that cannot be encoded" + ); + } + Err(e) => return Some(Err(e)), + }, Action::ResetHoldTimer(duration) => match &mut self.hold_timer { Some(timer) => timer.as_mut().reset(Instant::now() + duration), None => self.hold_timer = Some(Box::pin(sleep(duration))), @@ -351,50 +441,23 @@ impl Session { Action::StartKeepaliveTimer(every) => { // interval_at, so that the first tick is one interval // from now and not immediate. - self.keepalive_timer = Some(interval_at(Instant::now() + every, every)); + let mut timer = interval_at(Instant::now() + every, every); + // After a stall, one KEEPALIVE and not one for every + // tick that was missed. + timer.set_missed_tick_behavior(MissedTickBehavior::Delay); + self.keepalive_timer = Some(timer); } Action::Admit(negotiated) => { - self.ctx = ParserContext { - four_octet_asn: negotiated.four_octet_asn, - ibgp: negotiated.peer_asn == self.config.local_asn, - add_path: negotiated.add_path_receive.iter().collect(), - }; - self.send_ctx = ParserContext { - add_path: negotiated.add_path_send.iter().collect(), - ..self.ctx.clone() - }; - match self.config.collisions.admit( - self.config.router_id, - negotiated.peer_identifier, - self.config.direction, - ) { - Some(guard) => { - self.collision = Some(guard); - info!( - peer_addr = %self.peer_addr, - peer_asn = negotiated.peer_asn, - negotiated_hold_time = negotiated.hold_time_secs, - four_octet_asn = negotiated.four_octet_asn, - families = %negotiated.families, - add_path_receive = %negotiated.add_path_receive, - add_path_send = %negotiated.add_path_send, - direction = self.config.direction.as_str(), - "OPEN received, transitioning to OpenConfirm" - ); - } - None => { - // Not admitted: what the machine wanted next no - // longer applies. - info!(peer_addr = %self.peer_addr, "Closing connection: lost collision resolution"); - queue = self.fsm.handle(Event::CollisionLost).into(); - } + if !self.admit(&negotiated) { + // What the machine wanted next no longer applies. + queue = self.fsm.handle(Event::CollisionLost).into(); } } Action::Established(negotiated) => { - if let Err(full) = self.establish(negotiated) { + if let Err(e) = 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(); + error!(peer_addr = %self.peer_addr, err = %e, "Cannot take the peer's routes"); + queue = self.fsm.handle(Event::LocalFailure(e)).into(); } } Action::RoutesIgnored(count) => { @@ -404,34 +467,19 @@ impl Session { "Ignored routes of an address family that was not negotiated" ); } - Action::DeliverUpdate(update) => { - let attributes_wire = self.attributes_wire.take(); - if let Some(established) = &self.established { - 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())), - } + Action::DeliverUpdate { + update, + attributes_wire, + } => { + if let Err(e) = self.store(&update, attributes_wire.as_deref()) { + error!(peer_addr = %self.peer_addr, err = %e, "Cannot store the peer's routes"); + queue = self.fsm.handle(Event::LocalFailure(e)).into(); } } Action::RouteRefreshRequested => { if let Some(established) = &self.established { info!(peer = %established.peer_id, "Received ROUTE-REFRESH, re-sending our routes"); - self.config.registry.request_refresh(&established.peer_id); + self.server.registry.request_refresh(&established.peer_id); } } Action::Close(result) => return Some(result), @@ -440,18 +488,57 @@ impl Session { None } + /// Takes what the OPENs settled into use, and asks collision detection + /// whether this connection may go on. Returns whether it may. + fn admit(&mut self, negotiated: &Negotiated) -> bool { + self.ctx = ParserContext { + four_octet_asn: negotiated.four_octet_asn, + ibgp: negotiated.ibgp, + add_path: negotiated.add_path_receive.iter().collect(), + }; + self.send_ctx = ParserContext { + add_path: negotiated.add_path_send.iter().collect(), + ..self.ctx.clone() + }; + let hold_time = Duration::from_secs(negotiated.hold_time_secs.into()); + self.send_hold_time = MIN_SEND_HOLD_TIME.max(2 * hold_time); + + self.collision = self.server.collisions.admit( + self.server.router_id, + self.peer_addr.ip(), + negotiated.peer_identifier, + self.direction, + ); + if self.collision.is_none() { + info!(peer_addr = %self.peer_addr, "Closing connection: lost collision resolution"); + return false; + } + info!( + peer_addr = %self.peer_addr, + peer_asn = negotiated.peer_asn, + negotiated_hold_time = negotiated.hold_time_secs, + four_octet_asn = negotiated.four_octet_asn, + families = %negotiated.families, + add_path_receive = %negotiated.add_path_receive, + add_path_send = %negotiated.add_path_send, + direction = self.direction.as_str(), + "OPEN received, transitioning to OpenConfirm" + ); + true + } + /// 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> { + fn establish(&mut self, negotiated: &Negotiated) -> Result<(), SessionError> { let peer_id = PeerId { addr: self.peer_addr, asn: negotiated.peer_asn, }; - let rib_peer = self.config.rib.add_peer(PeerInfo { + let rib_peer = self.server.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, + ibgp: negotiated.ibgp, })?; info!(peer_addr = %self.peer_addr, peer_asn = negotiated.peer_asn, "Session established"); SESSIONS_ESTABLISHED_TOTAL.inc(); @@ -460,12 +547,12 @@ impl Session { collision.set_established(); } let (outbound_tx, outbound) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY); - let registration = self.config.registry.register(SessionHandle::new( + let registration = self.server.registry.register(SessionHandle::new( peer_id, - self.config.peer_name.clone(), - self.config.direction, + self.peer_name.clone(), + self.direction, self.local_addr, - &negotiated, + negotiated, rib_peer, outbound_tx, )); @@ -473,12 +560,40 @@ impl Session { peer_id, rib_peer, outbound, - routes: PeerRoutes::new(Arc::clone(&self.config.rib), rib_peer), + routes: PeerRoutes::new(Arc::clone(&self.server.rib), rib_peer), _registration: registration, }); Ok(()) } + /// Puts the routes of a received UPDATE into the routing table. + fn store( + &self, + update: &UpdateMessage, + attributes_wire: Option<&[u8]>, + ) -> Result<(), SessionError> { + let Some(established) = &self.established else { + return Ok(()); + }; + let outcome = routes::apply( + &self.server.rib, + established.rib_peer, + update, + attributes_wire, + )?; + // One line for every UPDATE is too much to log by default: a full + // table is a million of them. + debug!( + peer = %established.peer_id, + added = outcome.added, + replaced = outcome.replaced, + removed = outcome.removed, + best_changed = outcome.best_changed, + "Received UPDATE" + ); + 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); @@ -486,10 +601,14 @@ impl Session { } if let Message::Notification(notification) = msg { info!(peer_addr = %self.peer_addr, error = %notification.error, "Sending NOTIFICATION, closing session"); + self.notified = true; + } + self.write_buf.clear(); + msg.to_wire(&self.send_ctx, &mut self.write_buf)?; + match timeout(self.send_hold_time, self.writer.write_all(&self.write_buf)).await { + Ok(written) => written?, + Err(_) => return Err(SessionError::SendHoldTimerExpired(self.send_hold_time)), } - let mut buf = BytesMut::new(); - msg.to_wire(&self.send_ctx, &mut buf)?; - self.writer.write_all(&buf).await?; MESSAGES_SENT .with_label_values(&[message_type_name(msg)]) .inc(); @@ -533,11 +652,7 @@ async fn queued(established: &mut Option) -> Message { async fn killed(collision: &mut Option) { match collision { - Some(guard) => { - // An error means the collision table is gone, i.e. the server is - // going away; ending the session is right then too. - let _ = (&mut guard.killed).await; - } + Some(guard) => guard.killed().await, None => pending().await, } } @@ -592,6 +707,17 @@ impl FrameReader { } } } + + /// Reads and throws away whatever the peer sends, until it closes the + /// connection or the connection fails. + async fn discard_until_closed(&mut self) { + loop { + self.buf.clear(); + if !matches!(self.reader.read_buf(&mut self.buf).await, Ok(read) if read > 0) { + return; + } + } + } } /// The length of the message at the front of `buf`, once its header has @@ -721,20 +847,21 @@ mod tests { .await .unwrap(); let (stream, peer_addr) = listener.accept().await.unwrap(); - let config = SessionConfig { + let server = Arc::new(ServerContext { router_id: Ipv4Addr::new(10, 0, 0, 1), local_asn: 65000, - hold_time_secs: 90, rib: Arc::new(RibTable::new(4)), registry: SessionRegistry::new(), collisions: CollisionTable::new(false), - direction: Direction::Inbound, - peer_name: None, - expected_asn: None, - families: FamilySet::supported(), - add_path: None, - }; - let (mut session, _) = Session::open(stream, peer_addr, config).unwrap(); + }); + let (mut session, _) = Session::open( + stream, + peer_addr, + Direction::Inbound, + server, + PeerSettings::unconfigured(90), + ) + .unwrap(); let error = session .close_after_panic("index out of bounds".into()) @@ -753,6 +880,56 @@ mod tests { assert_eq!(received, expected); } + /// A peer that stops reading must not hold the session for as long as + /// TCP will keep trying: a write that makes no progress is given up + /// (RFC 9687). Nothing else would end it, since the session reads + /// nothing, timers included, while it waits to write. + #[tokio::test] + async fn a_peer_that_stops_reading_is_given_up_on() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + // Held, and never read from. + let _peer = TcpStream::connect(listener.local_addr().unwrap()) + .await + .unwrap(); + let (stream, peer_addr) = listener.accept().await.unwrap(); + let server = Arc::new(ServerContext { + router_id: Ipv4Addr::new(10, 0, 0, 1), + local_asn: 65000, + rib: Arc::new(RibTable::new(4)), + registry: SessionRegistry::new(), + collisions: CollisionTable::new(false), + }); + let (mut session, _) = Session::open( + stream, + peer_addr, + Direction::Inbound, + server, + PeerSettings::unconfigured(90), + ) + .unwrap(); + session.send_hold_time = Duration::from_millis(100); + + // Until the socket's buffers are full, writes succeed. + let filler = Message::Notification(NotificationMessage { + error: BgpError::Cease(CeaseSubcode::AdminShutdown), + data: vec![0; 4000], + }); + let stuck = tokio::time::timeout(Duration::from_secs(30), async { + loop { + if let Err(e) = session.send(&filler).await { + return e; + } + } + }) + .await + .expect("a write that cannot finish is given up"); + assert!( + matches!(stuck, SessionError::SendHoldTimerExpired(_)), + "{stuck}" + ); + assert!(stuck.notification().is_none(), "there is no way to say so"); + } + /// A session task that is torn down without running its cleanup must /// not leave the peer's routes behind. #[tokio::test] diff --git a/crates/server/src/dialer.rs b/crates/server/src/dialer.rs index 2100dde..c35a37b 100644 --- a/crates/server/src/dialer.rs +++ b/crates/server/src/dialer.rs @@ -17,8 +17,8 @@ use tracing::{info, warn}; use crate::collision::Direction; use crate::config::PeerConfig; -use crate::connection::{SessionConfig, run_session}; -use crate::session::{DialState, SessionRegistry}; +use crate::connection::{PeerSettings, ServerContext, run_session}; +use crate::session::DialState; /// Upper bound for exponential backoff between connection attempts. const MAX_BACKOFF: Duration = Duration::from_secs(120); @@ -26,23 +26,12 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); pub fn spawn_dialers( peers: &[PeerConfig], - base: &SessionConfig, + server: &Arc, connect_retry: Duration, ) -> Vec> { peers .iter() - .map(|peer| { - let config = SessionConfig { - hold_time_secs: peer.hold_time_secs, - direction: Direction::Outbound, - peer_name: Some(peer.name.clone()), - expected_asn: Some(peer.remote_asn), - families: peer.families.clone(), - add_path: peer.add_path, - ..base.clone() - }; - tokio::spawn(dial_loop(Arc::new(peer.clone()), config, connect_retry)) - }) + .map(|peer| tokio::spawn(dial_loop(peer.clone(), Arc::clone(server), connect_retry))) .collect() } @@ -64,8 +53,8 @@ async fn connect(peer: &PeerConfig) -> std::io::Result { socket.connect(peer.endpoint).await } -async fn dial_loop(peer: Arc, config: SessionConfig, connect_retry: Duration) { - let registry: SessionRegistry = config.registry.clone(); +async fn dial_loop(peer: PeerConfig, server: Arc, connect_retry: Duration) { + let registry = &server.registry; let mut backoff = connect_retry; loop { @@ -82,21 +71,32 @@ async fn dial_loop(peer: Arc, config: SessionConfig, connect_retry: registry.set_dial_state(&peer.name, DialState::Connecting); info!(peer = %peer.name, endpoint = %peer.endpoint, "Connecting to peer"); - let outcome = match tokio::time::timeout(CONNECT_TIMEOUT, connect(&peer)).await { + let ended = match tokio::time::timeout(CONNECT_TIMEOUT, connect(&peer)).await { Ok(Ok(stream)) => { info!(peer = %peer.name, endpoint = %peer.endpoint, "TCP connected, starting session"); - backoff = connect_retry; - run_session(stream, peer.endpoint, config.clone()) - .await - .map(|()| "session closed by peer".to_string()) - .map_err(|e| e.to_string()) + let end = run_session( + stream, + peer.endpoint, + Direction::Outbound, + Arc::clone(&server), + PeerSettings::from(&peer), + ) + .await; + // A session that came up is a peer worth dialling again + // soon. One that connects and is then refused, for its AS + // number say, will be refused again: it is retried ever + // more slowly, like one that cannot be reached at all. + if end.established { + backoff = connect_retry; + } + end.result.map_err(|e| e.to_string()) } Ok(Err(e)) => Err(format!("connect failed: {e}")), Err(_) => Err(format!("connect timed out after {CONNECT_TIMEOUT:?}")), }; - let reason = match outcome { - Ok(msg) => msg, + let reason = match ended { + Ok(()) => "session closed at our request".to_string(), Err(reason) => { warn!(peer = %peer.name, endpoint = %peer.endpoint, err = %reason, "Session attempt ended"); reason diff --git a/crates/server/src/error.rs b/crates/server/src/error.rs index 9639006..e20a3cd 100644 --- a/crates/server/src/error.rs +++ b/crates/server/src/error.rs @@ -2,6 +2,7 @@ use std::io; use std::net::SocketAddr; +use std::time::Duration; use bgp_packet::errors::{BgpError, CeaseSubcode, DecodeError, FsmSubcode, OpenMsgSubcode}; use bgp_packet::message::NotificationMessage; @@ -41,8 +42,6 @@ pub enum SessionError { WrongPeerAs { expected: u32, actual: u32 }, #[error("closed by connection collision resolution")] Collision, - /// The peer offers none of the address families configured for it, so - /// the session could carry nothing. /// The routing table has no room for another peer. #[error(transparent)] TableFull(#[from] TooManyPeers), @@ -50,8 +49,14 @@ pub enum SessionError { /// is being reset, and the session is cleaned up like any other. #[error("the session panicked: {0}")] Panicked(String), + /// The peer offers none of the address families configured for it, so + /// the session could carry nothing. #[error("no address family in common: we offer {ours}, the peer offers {theirs}")] NoCommonAddressFamily { ours: FamilySet, theirs: FamilySet }, + /// The peer took nothing we sent for this long (RFC 9687): it has + /// stopped reading, and there is no telling it so. + #[error("the peer accepted no data for {0:?}")] + SendHoldTimerExpired(Duration), } impl SessionError { @@ -59,7 +64,10 @@ impl SessionError { /// or `None` when there is nothing to say or no way to say it. pub fn notification(&self) -> Option { let error = match self { - Self::Io(_) | Self::Encode(_) | Self::NotifiedByPeer(_) => return None, + Self::Io(_) | Self::SendHoldTimerExpired(_) | Self::NotifiedByPeer(_) => return None, + // Something of ours could not be written out, which is a bug + // here: as for a panic, the session is reset. + Self::Encode(_) => BgpError::Cease(CeaseSubcode::AdminReset), Self::Decode(e) => { return Some(NotificationMessage { error: e.notification.clone(), @@ -97,7 +105,7 @@ impl SessionError { } } -/// Why the server could not start. +/// Why the server could not start, or could not go on. #[derive(Debug, Error)] pub enum ServerError { #[error("failed to bind listener on {addr}: {source}")] @@ -120,7 +128,7 @@ pub enum ServerError { /// Why a configuration was rejected. #[derive(Debug, Clone, PartialEq, Eq, Error)] pub enum ConfigError { - #[error("invalid router_id {0:?}: must be a dotted-quad IPv4 address")] + #[error("invalid router_id {0:?}: must be a dotted-quad IPv4 address other than 0.0.0.0")] RouterId(String), #[error("local_asn must be non-zero")] LocalAsn, @@ -150,6 +158,10 @@ pub enum PeerProblem { RemoteAsn, #[error("hold_time_seconds must be 0 or at least 3")] HoldTime, + #[error("there is another peer of this name")] + DuplicateName, + #[error("there is another peer at this address")] + DuplicateAddress, #[error("local_address {0:?} is not an IP address")] LocalAddress(String), #[error("local_address and endpoint are different address families")] @@ -217,6 +229,16 @@ mod tests { }), Some((BgpError::OpenMsg(OpenMsgSubcode::BadPeerAs), vec![])) ); + // Something of ours that cannot be written is our fault: the peer + // is told the session is being reset, not left to time it out. + assert_eq!( + notification(SessionError::Encode(EncodeError::TooLong { + what: "path attribute", + len: 70000, + max: 65535 + })), + Some((BgpError::Cease(CeaseSubcode::AdminReset), vec![])) + ); // A broken connection cannot carry a NOTIFICATION. assert_eq!( notification(SessionError::Io(io::ErrorKind::BrokenPipe.into())), diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index 2f9a737..a616da8 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -18,8 +18,6 @@ use bgp_packet::errors::{BgpError, DecodeError, OpenMsgSubcode}; use bgp_packet::message::{AS_TRANS, BGP4_VERSION, Message, UpdateMessage}; use bgp_packet::open::{AddPathEntry, AddPathSendRecv, Capability, OpenMessage, OpenOption}; -use bgp_rib::table::TooManyPeers; - use crate::error::SessionError; use crate::family::FamilySet; use crate::metrics::message_type_name; @@ -73,6 +71,8 @@ pub struct FsmConfig { #[derive(Debug, Clone, PartialEq, Eq)] pub struct Negotiated { pub peer_asn: u32, + /// Whether the peer is in our own AS. + pub ibgp: bool, pub peer_identifier: Ipv4Addr, /// Zero means the hold and keepalive timers do not run (RFC 4271 §4.2). pub hold_time_secs: u16, @@ -90,11 +90,30 @@ pub struct Negotiated { pub add_path_send: FamilySet, } +/// A message from the peer, with what the decoder found out besides. +#[derive(Debug)] +pub struct Received { + pub message: Message, + /// For an UPDATE: its attributes as they were on the wire, if the + /// decoder kept them (`DecodedUpdate::attributes_wire`). They go with + /// the UPDATE to whoever stores its routes. + pub attributes_wire: Option>, +} + +impl From for Received { + fn from(message: Message) -> Self { + Self { + message, + attributes_wire: None, + } + } +} + /// Something that happened to the connection. #[derive(Debug)] pub enum Event { /// The peer sent a message. - Message(Message), + Message(Received), /// The peer sent something that is not a valid message. DecodeFailed(DecodeError), HoldTimerExpired, @@ -104,9 +123,10 @@ pub enum Event { CollisionLost, /// The rest of the server wants this message sent to the peer. Outbound(Message), - /// The routing table has no room for this peer's routes, so the session - /// cannot go on. - TableFull(TooManyPeers), + /// Something on our side keeps the session from going on: the routing + /// table has no room for the peer's routes, say. The peer is told as + /// the error has it. + LocalFailure(SessionError), } /// Something the driver must do, in the order given. @@ -128,8 +148,12 @@ pub enum Action { /// The peer sent this many routes of address families that were not /// negotiated. They have been left out of the update that follows. RoutesIgnored(usize), - /// Routes arrived; hand them to the RIB. - DeliverUpdate(UpdateMessage), + /// Routes arrived; put them in the routing table. + DeliverUpdate { + update: UpdateMessage, + /// See [`Received::attributes_wire`]. + attributes_wire: Option>, + }, /// The peer asked for everything we advertise to be sent again. RouteRefreshRequested, /// The session is over, for this reason. Always the last action. @@ -179,7 +203,8 @@ impl Fsm { } /// What the OPENs settled; `None` until the peer's OPEN is accepted. - pub fn negotiated(&self) -> Option<&Negotiated> { + #[cfg(test)] + fn negotiated(&self) -> Option<&Negotiated> { match &self.phase { Phase::OpenConfirm(negotiated) | Phase::Established(negotiated) => Some(negotiated), Phase::OpenSent | Phase::Closed => None, @@ -192,10 +217,10 @@ impl Fsm { return vec![]; } match event { - Event::Message(msg) => self.on_message(msg), + Event::Message(received) => self.on_message(received), Event::DecodeFailed(e) => self.fail(SessionError::Decode(e)), Event::CollisionLost => self.fail(SessionError::Collision), - Event::TableFull(e) => self.fail(SessionError::TableFull(e)), + Event::LocalFailure(e) => self.fail(e), Event::HoldTimerExpired => self.fail(match self.phase { Phase::Established(_) => SessionError::HoldTimerExpired, _ => SessionError::Timeout(self.state()), @@ -208,8 +233,12 @@ impl Fsm { } } - fn on_message(&mut self, msg: Message) -> Vec { - match (&self.phase, msg) { + fn on_message(&mut self, received: Received) -> Vec { + let Received { + message, + attributes_wire, + } = received; + match (&self.phase, message) { // A NOTIFICATION ends the session in any state, and is not // answered (RFC 4271 §8.2.2). (_, Message::Notification(notification)) => { @@ -240,7 +269,10 @@ impl Fsm { if ignored > 0 { actions.push(Action::RoutesIgnored(ignored)); } - actions.push(Action::DeliverUpdate(update)); + actions.push(Action::DeliverUpdate { + update, + attributes_wire, + }); actions } (Phase::Established(negotiated), Message::RouteRefresh(_)) => { @@ -304,6 +336,7 @@ impl Fsm { let negotiated = Negotiated { peer_asn, + ibgp: peer_asn == self.config.local_asn, peer_identifier: open.identifier, hold_time_secs: self.config.hold_time_secs.min(open.hold_time), four_octet_asn: capability_asn.is_some(), @@ -428,6 +461,8 @@ mod tests { use bgp_primitives::constants::AfiSafi; use bgp_primitives::ip_prefix::IpPrefix; + use bgp_rib::table::TooManyPeers; + use super::*; use crate::family::SUPPORTED; @@ -470,15 +505,19 @@ mod tests { fn fsm_in(state: State, peer_hold_time: u16) -> Fsm { let (mut fsm, _) = Fsm::new(config()); if state != State::OpenSent { - fsm.handle(Event::Message(Message::Open(peer_open(peer_hold_time)))); + fsm.handle(received(Message::Open(peer_open(peer_hold_time)))); } if state == State::Established { - fsm.handle(Event::Message(Message::KeepAlive)); + fsm.handle(received(Message::KeepAlive)); } assert_eq!(fsm.state(), state); fsm } + fn received(message: Message) -> Event { + Event::Message(message.into()) + } + fn empty_update() -> UpdateMessage { UpdateMessage { withdrawn_routes: vec![], @@ -525,10 +564,11 @@ mod tests { let (mut fsm, _) = Fsm::new(config()); // The peer proposes 30 seconds, we 90: the smaller one holds. - let actions = fsm.handle(Event::Message(Message::Open(peer_open(30)))); + let actions = fsm.handle(received(Message::Open(peer_open(30)))); assert_eq!(fsm.state(), State::OpenConfirm); let expected = Negotiated { peer_asn: 65001, + ibgp: false, peer_identifier: PEER_ID, hold_time_secs: 30, four_octet_asn: false, @@ -546,7 +586,7 @@ mod tests { ] if *n == expected && *hold == Duration::from_secs(30) )); - let actions = fsm.handle(Event::Message(Message::KeepAlive)); + let actions = fsm.handle(received(Message::KeepAlive)); assert_eq!(fsm.state(), State::Established); assert!(matches!( &actions[..], @@ -563,20 +603,20 @@ mod tests { #[test] fn a_hold_time_of_zero_runs_no_timers_once_established() { let (mut fsm, _) = Fsm::new(config()); - let actions = fsm.handle(Event::Message(Message::Open(peer_open(0)))); + let actions = fsm.handle(received(Message::Open(peer_open(0)))); // The KEEPALIVE that confirms the session is still waited for. assert!(matches!( actions.last(), Some(Action::ResetHoldTimer(INITIAL_HOLD_TIME)) )); - let actions = fsm.handle(Event::Message(Message::KeepAlive)); + let actions = fsm.handle(received(Message::KeepAlive)); assert!(matches!( &actions[..], [Action::Established(_), Action::StopHoldTimer] )); // And hearing from the peer restarts nothing. - assert!(fsm.handle(Event::Message(Message::KeepAlive)).is_empty()); + assert!(fsm.handle(received(Message::KeepAlive)).is_empty()); } #[test] @@ -592,7 +632,7 @@ mod tests { }])], ..peer_open(90) }; - fsm.handle(Event::Message(Message::Open(open))); + fsm.handle(received(Message::Open(open))); let negotiated = fsm.negotiated().unwrap(); assert_eq!(negotiated.peer_asn, 210036); assert!(negotiated.four_octet_asn); @@ -630,7 +670,7 @@ mod tests { )], ..peer_open(90) }; - let actions = fsm.handle(Event::Message(Message::Open(open))); + let actions = fsm.handle(received(Message::Open(open))); (fsm, actions) }; @@ -691,7 +731,7 @@ mod tests { options: vec![OpenOption::Capabilities(capabilities)], ..peer_open(90) }; - fsm.handle(Event::Message(Message::Open(open))); + fsm.handle(received(Message::Open(open))); let n = fsm.negotiated().unwrap().clone(); (first, n.add_path_receive, n.add_path_send) }; @@ -754,14 +794,14 @@ mod tests { announced_routes: vec![v4.into()], }; - let actions = fsm.handle(Event::Message(Message::Update(update))); + let actions = fsm.handle(received(Message::Update(update))); assert_eq!(fsm.state(), State::Established, "the session survives"); assert!(matches!( &actions[..], [ Action::ResetHoldTimer(_), Action::RoutesIgnored(1), - Action::DeliverUpdate(delivered) + Action::DeliverUpdate { update: delivered, .. } ] if delivered.announced_routes == [v4] && delivered.path_attributes.is_empty() )); } @@ -770,7 +810,7 @@ mod tests { fn an_unacceptable_open_is_refused_with_the_matching_subcode() { let refused = |open: OpenMessage, config: FsmConfig| { let (mut fsm, _) = Fsm::new(config); - let actions = fsm.handle(Event::Message(Message::Open(open))); + let actions = fsm.handle(received(Message::Open(open))); ending(&fsm, actions).0 }; let open_error = |subcode| Some(BgpError::OpenMsg(subcode)); @@ -833,7 +873,7 @@ mod tests { ]; for (state, msg, subcode) in cases { let mut fsm = fsm_in(state, 90); - let actions = fsm.handle(Event::Message(msg)); + let actions = fsm.handle(received(msg)); let (sent, result) = ending(&fsm, actions); assert_eq!(sent, Some(BgpError::FsmError(subcode)), "{state}"); assert!( @@ -899,7 +939,7 @@ mod tests { for state in [State::OpenSent, State::OpenConfirm, State::Established] { let mut fsm = fsm_in(state, 90); let cease = BgpError::Cease(CeaseSubcode::AdminShutdown); - let actions = fsm.handle(Event::Message(Message::Notification(NotificationMessage { + let actions = fsm.handle(received(Message::Notification(NotificationMessage { error: cease.clone(), data: vec![], }))); @@ -914,16 +954,16 @@ mod tests { let mut fsm = fsm_in(State::Established, 30); let hold = Duration::from_secs(30); - let actions = fsm.handle(Event::Message(Message::KeepAlive)); + let actions = fsm.handle(received(Message::KeepAlive)); assert!(matches!(&actions[..], [Action::ResetHoldTimer(d)] if *d == hold)); - let actions = fsm.handle(Event::Message(Message::Update(empty_update()))); + let actions = fsm.handle(received(Message::Update(empty_update()))); assert!(matches!( &actions[..], - [Action::ResetHoldTimer(d), Action::DeliverUpdate(_)] if *d == hold + [Action::ResetHoldTimer(d), Action::DeliverUpdate { .. }] if *d == hold )); - let actions = fsm.handle(Event::Message(Message::RouteRefresh( + let actions = fsm.handle(received(Message::RouteRefresh( RouteRefreshMessage::request(AfiSafi::IPV4_UNICAST), ))); assert!(matches!( @@ -970,7 +1010,7 @@ mod tests { #[test] fn a_full_table_ends_the_session_and_says_why() { let mut fsm = fsm_in(State::Established, 90); - let actions = fsm.handle(Event::TableFull(TooManyPeers)); + let actions = fsm.handle(Event::LocalFailure(SessionError::TableFull(TooManyPeers))); let (sent, result) = ending(&fsm, actions); assert_eq!(sent, Some(BgpError::Cease(CeaseSubcode::OutOfResources))); assert!(matches!(result, Err(SessionError::TableFull(_)))); @@ -982,7 +1022,7 @@ mod tests { fsm.handle(Event::HoldTimerExpired); assert_eq!(fsm.state(), State::Closed); for event in [ - Event::Message(Message::KeepAlive), + received(Message::KeepAlive), Event::HoldTimerExpired, Event::KeepaliveTimerExpired, Event::CollisionLost, diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index 96a6419..bd0f879 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -1,3 +1,6 @@ +//! The server: listeners, the tasks that run for as long as it does, and +//! what happens when one of them stops. + use std::io; use std::net::SocketAddr; use std::pin::Pin; @@ -5,7 +8,7 @@ use std::sync::Arc; use std::time::Duration; use futures::stream::{FuturesUnordered, StreamExt}; -use tokio::net::{TcpListener, TcpStream}; +use tokio::net::{TcpListener, TcpSocket, TcpStream}; use tokio::task::{JoinError, JoinHandle}; use tracing::{error, info, warn}; @@ -15,19 +18,30 @@ use bgp_rib::{LocalRib, RibStopped}; use crate::advertise::spawn_advertiser; use crate::collision::{CollisionTable, Direction}; use crate::config::Config; -use crate::connection::{SessionConfig, handle_connection}; +use crate::connection::{PeerSettings, ServerContext, run_session}; use crate::dialer::spawn_dialers; use crate::error::ServerError; -use crate::family::FamilySet; use crate::panic; use crate::session::{ConfiguredPeer, SessionRegistry}; -pub struct BgpServerBuilder { - config: Config, -} +/// How long sessions get to send their Cease when the server stops. +const SHUTDOWN_GRACE: Duration = Duration::from_secs(2); -/// A `BgpServerBuilder` that has bound all its listeners and is ready to accept. -/// Call [`BgpServer::run`] to start the accept loop. +/// How long to wait before accepting again after an accept failed. What +/// makes it fail (no file descriptors left, usually) does not pass at once, +/// and trying again in a loop would only fill the log. +const ACCEPT_RETRY: Duration = Duration::from_millis(200); + +/// How often attribute sets that no route uses any more are dropped. They +/// are not dropped as their last route goes (see `AttrStore::sweep`), and a +/// table whose routes keep changing attributes collects them. +const SWEEP_INTERVAL: Duration = Duration::from_secs(60); + +/// How many connections may wait to be accepted on a listener. +const LISTEN_BACKLOG: u32 = 1024; + +/// A BGP speaker that has bound its listeners and is ready to accept. +/// Call [`BgpServer::run`] to start it. pub struct BgpServer { config: Config, listeners: Vec<(SocketAddr, TcpListener)>, @@ -41,20 +55,14 @@ pub struct BgpServer { pub registry: SessionRegistry, } -impl BgpServerBuilder { - pub fn new(config: Config) -> Self { - Self { config } - } - - /// Binds all configured listen endpoints and returns a [`BgpServer`]. - /// Separating bind from run lets callers (e.g. tests) inspect the actual - /// bound addresses before the accept loop starts. - pub async fn bind(self) -> Result { - let mut listeners = Vec::with_capacity(self.config.listen_endpoints.len()); - for &addr in &self.config.listen_endpoints { - let listener = TcpListener::bind(addr) - .await - .map_err(|source| ServerError::Bind { addr, source })?; +impl BgpServer { + /// Binds all configured listen endpoints. Separate from + /// [`BgpServer::run`] so that callers (tests, say) can learn the + /// addresses actually bound before any connection is accepted. + pub async fn bind(config: Config) -> Result { + let mut listeners = Vec::with_capacity(config.listen_endpoints.len()); + for &addr in &config.listen_endpoints { + let listener = listen(addr).map_err(|source| ServerError::Bind { addr, source })?; let bound_addr = listener.local_addr()?; info!(%bound_addr, "Listening for BGP connections"); listeners.push((bound_addr, listener)); @@ -62,35 +70,27 @@ impl BgpServerBuilder { // Routes from the configuration go in before any session exists; // each session is sent the whole table when it comes up. let local_rib = LocalRib::spawn(); - for route in &self.config.originate { + for route in &config.originate { local_rib.announce(route.clone()).await?; } + let configured = config + .peers + .iter() + .map(|p| ConfiguredPeer { + name: p.name.clone(), + endpoint: p.endpoint, + remote_asn: p.remote_asn, + }) + .collect(); Ok(BgpServer { listeners, rib: Arc::new(RibTable::new(table_shards())), local_rib, - registry: SessionRegistry::with_configured( - self.config - .peers - .iter() - .map(|p| ConfiguredPeer { - name: p.name.clone(), - endpoint: p.endpoint, - remote_asn: p.remote_asn, - }) - .collect(), - ), - config: self.config, + registry: SessionRegistry::with_configured(configured), + config, }) } - /// Convenience method: bind then run immediately. - pub async fn run(self) -> Result<(), ServerError> { - self.bind().await?.run().await - } -} - -impl BgpServer { pub fn config(&self) -> &Config { &self.config } @@ -100,6 +100,8 @@ impl BgpServer { self.listeners.iter().map(|(addr, _)| *addr).collect() } + /// Runs the server. Returns only if something it cannot do without has + /// failed, after closing its sessions. pub async fn run(self) -> Result<(), ServerError> { crate::metrics::init(); info!( @@ -109,30 +111,18 @@ impl BgpServer { "BGP server started" ); - let session_config = SessionConfig { + let context = Arc::new(ServerContext { router_id: self.config.router_id, local_asn: self.config.local_asn, - hold_time_secs: self.config.hold_time_secs, rib: Arc::clone(&self.rib), registry: self.registry.clone(), collisions: CollisionTable::new(self.config.collision_detect_established_state), - direction: Direction::Inbound, - peer_name: None, - expected_asn: None, - families: FamilySet::supported(), - add_path: None, - }; + }); // The tasks the server cannot do without. None of them ends while - // the server runs, unless it panics. - let mut tasks: FuturesUnordered<_> = spawn_dialers( - &self.config.peers, - &session_config, - self.config.connect_retry, - ) - .into_iter() - .map(|dialer| named("dialer", dialer)) - .collect(); + // the server runs. The advertiser comes first: it has to be + // listening before a dialer brings up the first session. + let mut tasks: FuturesUnordered = FuturesUnordered::new(); tasks.push(named( "advertiser", spawn_advertiser( @@ -141,6 +131,10 @@ impl BgpServer { self.config.local_asn, ), )); + tasks.push(named("sweeper", spawn_sweeper(Arc::clone(&self.rib)))); + for dialer in spawn_dialers(&self.config.peers, &context, self.config.connect_retry) { + tasks.push(named("dialer", dialer)); + } let mut accepting: FuturesUnordered<_> = self .listeners @@ -152,31 +146,24 @@ impl BgpServer { let (listen_addr, result, listener) = tokio::select! { Some(accepted) = accepting.next() => accepted, failure = failed(&self.local_rib, &mut tasks) => { - return Err(give_up(&self.registry, failure).await); + error!(err = %failure, "Shutting down: the server cannot continue"); + close_sessions(&self.registry).await; + return Err(failure); } }; match result { Ok((stream, peer_addr)) => { - // Inbound connections from a configured peer inherit its - // name, expected ASN, hold time, address families and - // ADD-PATH setting. - let mut config = session_config.clone(); - if let Some(peer) = self - .config - .peers - .iter() - .find(|p| p.endpoint.ip() == peer_addr.ip()) - { - config.peer_name = Some(peer.name.clone()); - config.expected_asn = Some(peer.remote_asn); - config.hold_time_secs = peer.hold_time_secs; - config.families = peer.families.clone(); - config.add_path = peer.add_path; - } - tokio::spawn(handle_connection(stream, peer_addr, Arc::new(config))); + // A connection from a configured peer is run with what + // is configured for that peer. + let peer = self.config.peer_at(peer_addr.ip()).map_or_else( + || PeerSettings::unconfigured(self.config.hold_time_secs), + PeerSettings::from, + ); + tokio::spawn(serve(stream, peer_addr, Arc::clone(&context), peer)); } Err(e) => { error!(%listen_addr, err = %e, "Accept error"); + tokio::time::sleep(ACCEPT_RETRY).await; } } accepting.push(accept_one(listen_addr, listener)); @@ -184,17 +171,60 @@ impl BgpServer { } } -/// Ends the server because a part of it that it cannot do without has -/// failed. Peers are told we are going (Cease / Administrative Shutdown), -/// so that they stop using our routes now and not when their hold timers -/// expire. -async fn give_up(registry: &SessionRegistry, failure: ServerError) -> ServerError { - error!(err = %failure, "Shutting down: the server cannot continue"); +/// Closes every session politely and waits a moment for them to go: each +/// peer is told we are going (Cease / Administrative Shutdown), so that it +/// stops using our routes now and not when its hold timer expires. +pub async fn close_sessions(registry: &SessionRegistry) { let left = registry.shutdown(SHUTDOWN_GRACE).await; if left > 0 { warn!(sessions = left, "Some sessions did not close in time"); } - failure +} + +/// Opens a listening socket on `addr`. +fn listen(addr: SocketAddr) -> io::Result { + let socket = match addr { + SocketAddr::V4(_) => TcpSocket::new_v4()?, + SocketAddr::V6(_) => { + let socket = TcpSocket::new_v6()?; + // IPv6 only. Where a system's default is to take IPv4 + // connections on such a socket too, listening on both 0.0.0.0 + // and [::] would fail on the second, and IPv4 peers would come + // in under IPv4-mapped addresses that match no configured peer. + socket2::SockRef::from(&socket).set_only_v6(true)?; + socket + } + }; + socket.set_reuseaddr(true)?; + socket.bind(addr)?; + socket.listen(LISTEN_BACKLOG) +} + +/// Runs a session on a connection a peer opened, and logs how it ended. +async fn serve( + stream: TcpStream, + peer_addr: SocketAddr, + context: Arc, + peer: PeerSettings, +) { + let end = run_session(stream, peer_addr, Direction::Inbound, context, peer).await; + if let Err(e) = end.result { + warn!(%peer_addr, err = %e, "BGP session terminated with error"); + } +} + +/// Drops, every so often, the attribute sets no route uses any more. +fn spawn_sweeper(rib: Arc) -> JoinHandle<()> { + tokio::spawn(async move { + let mut tick = tokio::time::interval(SWEEP_INTERVAL); + loop { + tick.tick().await; + let rib = Arc::clone(&rib); + // Takes each of the store's locks in turn and looks at every + // set: not for a thread that sessions run on. + let _ = tokio::task::spawn_blocking(move || rib.attrs().sweep()).await; + } + }) } /// How many shards the routing table has. Sessions write to it from every @@ -204,9 +234,6 @@ 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); - type NamedTask = Pin)> + Send>>; fn named(name: &'static str, task: JoinHandle<()>) -> NamedTask { diff --git a/crates/server/src/session.rs b/crates/server/src/session.rs index c4030e5..5815d04 100644 --- a/crates/server/src/session.rs +++ b/crates/server/src/session.rs @@ -15,7 +15,6 @@ use bgp_packet::errors::{BgpError, CeaseSubcode}; use bgp_packet::message::NotificationMessage; use bgp_packet::message::Message; -use bgp_rib::PeerId; use bgp_rib::table::PeerIndex; use tokio::sync::{broadcast, mpsc}; @@ -27,6 +26,19 @@ use crate::sync::IgnorePoisonRw; const EVENT_CHANNEL_CAPACITY: usize = 256; +/// Identifies a session by the peer's TCP address and its AS number. +#[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) + } +} + /// Where the dialer is in bringing up a configured peer. #[derive(Debug, Clone, PartialEq, Eq)] pub enum DialState { @@ -36,6 +48,7 @@ pub enum DialState { /// The last attempt failed or the session ended; waiting to retry. Backoff { until: SystemTime, + /// How the last attempt or session ended. last_error: String, }, } @@ -60,6 +73,8 @@ pub struct SessionHandle { /// originate to this peer. pub local_addr: SocketAddr, pub established_at: SystemTime, + /// Whether the peer is in our own AS. + pub ibgp: bool, /// Whether the peer negotiated RFC 6793 four-octet ASNs. Determines how /// AS_PATH is encoded on the wire for this session. pub four_octet_asn: bool, @@ -82,7 +97,7 @@ pub struct SessionHandle { pub struct SessionClosed(pub PeerId); impl SessionHandle { - pub fn new( + pub(crate) fn new( peer: PeerId, name: Option, direction: Direction, @@ -97,6 +112,7 @@ impl SessionHandle { direction, local_addr, established_at: SystemTime::now(), + ibgp: negotiated.ibgp, four_octet_asn: negotiated.four_octet_asn, families: negotiated.families.clone(), add_path_receive: negotiated.add_path_receive.clone(), @@ -176,12 +192,7 @@ impl SessionRegistry { .collect() } - /// The configured peer whose endpoint IP is `ip`, if any. - pub fn configured_by_ip(&self, ip: IpAddr) -> Option<&ConfiguredPeer> { - self.configured.iter().find(|p| p.endpoint.ip() == ip) - } - - pub fn set_dial_state(&self, name: &str, state: DialState) { + pub(crate) fn set_dial_state(&self, name: &str, state: DialState) { self.dial_states.writing().insert(name.to_string(), state); } @@ -197,7 +208,7 @@ impl SessionRegistry { /// Adds a session and emits [`SessionEvent::Up`]. The returned guard /// removes the session again when dropped. - pub fn register(&self, handle: SessionHandle) -> SessionGuard { + pub(crate) fn register(&self, handle: SessionHandle) -> SessionGuard { let peer = handle.peer; self.inner.writing().insert(peer, handle.clone()); SESSIONS_ESTABLISHED.inc(); @@ -208,18 +219,15 @@ impl SessionRegistry { } } - pub fn get(&self, peer: &PeerId) -> Option { - self.inner.reading().get(peer).cloned() - } - /// All currently established sessions, in no particular order. pub fn list(&self) -> Vec { self.inner.reading().values().cloned().collect() } /// Called by a session when its peer asks for a route refresh. - pub fn request_refresh(&self, peer: &PeerId) { - if let Some(handle) = self.get(peer) { + pub(crate) fn request_refresh(&self, peer: &PeerId) { + let handle = self.inner.reading().get(peer).cloned(); + if let Some(handle) = handle { let _ = self.events.send(SessionEvent::RefreshRequested(handle)); } } @@ -237,12 +245,15 @@ impl SessionRegistry { pub async fn shutdown(&self, grace: Duration) -> usize { self.shutting_down.store(true, Ordering::Relaxed); for session in self.list() { + // Not waited for: a session whose queue is full is one whose + // peer is not reading, and it would hold up the goodbye to all + // the others. It ends when the process does. let _ = session - .send(Message::Notification(NotificationMessage { + .outbound + .try_send(Message::Notification(NotificationMessage { error: BgpError::Cease(CeaseSubcode::AdminShutdown), data: vec![], - })) - .await; + })); } let deadline = tokio::time::Instant::now() + grace; while !self.list().is_empty() && tokio::time::Instant::now() < deadline { diff --git a/crates/server/src/tracing.rs b/crates/server/src/tracing.rs index 5139809..afaa7de 100644 --- a/crates/server/src/tracing.rs +++ b/crates/server/src/tracing.rs @@ -8,12 +8,10 @@ const DEFAULT_TARGETS: [&str; 3] = ["bgp", "bgp_server", "bgp_web"]; #[derive(Debug, Args)] pub struct TracingArgs { - /// Set the minimum log level. - #[arg(short, long, global = true, default_value_t = Level::TRACE, help_heading = "Tracing")] + /// Set the minimum log level. At `debug` and below there is a line for + /// every UPDATE and every route announced. + #[arg(short, long, global = true, default_value_t = Level::INFO, help_heading = "Tracing")] level: Level, - /// Disable logging to stdout. - #[arg(long, global = true, default_value_t = false, help_heading = "Tracing")] - quiet: bool, } impl TracingArgs { diff --git a/crates/web/src/api.rs b/crates/web/src/api.rs index f2dfd29..5ca02dc 100644 --- a/crates/web/src/api.rs +++ b/crates/web/src/api.rs @@ -12,8 +12,8 @@ use axum::response::{IntoResponse, Response}; use bgp_packet::community; use bgp_packet::message::{NlriNextHop, OriginPathAttribute, PathAttribute}; use bgp_rib::table::{PeerIndex, Route}; -use bgp_rib::{LocalRoute, PeerId, RibStopped}; -use bgp_server::session::{ConfiguredPeer, DialState, SessionHandle, SessionRegistry}; +use bgp_rib::{LocalRoute, RibStopped}; +use bgp_server::session::{ConfiguredPeer, DialState, PeerId, SessionHandle, SessionRegistry}; use serde::{Deserialize, Serialize}; use crate::AppState; diff --git a/crates/web/src/events.rs b/crates/web/src/events.rs index 050c5b3..c1ffe3c 100644 --- a/crates/web/src/events.rs +++ b/crates/web/src/events.rs @@ -8,7 +8,7 @@ use axum::extract::State; use axum::response::sse::{Event, KeepAlive, Sse}; use bgp_primitives::ip_prefix::IpPrefix; use bgp_rib::LocalRibEvent; -use bgp_rib::table::{self as journal, Route}; +use bgp_rib::table::{self, Route}; use bgp_server::session::SessionEvent; use futures::stream::{Stream, StreamExt}; use serde::Serialize; @@ -116,12 +116,12 @@ fn best_routes(state: &AppState) -> impl Stream + use<> { // the views are made afterwards. let mut changes = Vec::new(); let progress = reader.poll(BEST_ROUTE_BUFFER, |event| match event { - journal::Event::Reset => { + table::Event::Reset => { syncing = true; changes.push(Change::Lagged); } - journal::Event::Synced => syncing = false, - journal::Event::Best { + table::Event::Synced => syncing = false, + table::Event::Best { prefix, peer, path_id, @@ -135,7 +135,7 @@ fn best_routes(state: &AppState) -> impl Stream + use<> { best: true, })); } - journal::Event::Withdrawn { prefix } if !syncing => { + table::Event::Withdrawn { prefix } if !syncing => { changes.push(Change::Withdrawn(prefix)); } _ => {} diff --git a/crates/web/src/lib.rs b/crates/web/src/lib.rs index 7004ff8..5e65f24 100644 --- a/crates/web/src/lib.rs +++ b/crates/web/src/lib.rs @@ -2,7 +2,7 @@ //! status page, a server-sent event stream, and Prometheus metrics. //! //! Everything here is a thin layer over handles owned by the BGP server -//! ([`SessionRegistry`], [`LocalRib`], [`AdjRibIn`]); no routing state lives +//! ([`SessionRegistry`], [`LocalRib`], [`RibTable`]); no routing state lives //! in this crate. mod api;