diff --git a/crates/server/src/advertise.rs b/crates/server/src/advertise.rs index 83e5230..930c24a 100644 --- a/crates/server/src/advertise.rs +++ b/crates/server/src/advertise.rs @@ -166,7 +166,7 @@ impl Advertiser { session: session.clone(), local_asn: self.local_asn, resend: Arc::clone(&resend), - announced: HashSet::new(), + own_announced: HashSet::new(), end_of_rib: session .graceful_restart .as_ref() @@ -215,9 +215,10 @@ struct Exporter { session: SessionHandle, local_asn: u32, resend: Arc, - /// The prefixes announced to the peer and not withdrawn since: our - /// Adj-RIB-Out for it (RFC 4271 §3.2), as far as it needs keeping. - announced: HashSet, + /// The prefixes of our own routes announced to the peer and not + /// withdrawn since: our Adj-RIB-Out for it (RFC 4271 §3.2), as far as + /// it needs keeping. The learned ones are in `learned`. + own_announced: HashSet, /// The families to send End-of-RIB markers for, until they are sent; /// `None` if graceful restart was not agreed. end_of_rib: Option, @@ -271,12 +272,16 @@ struct Dressed { octets: usize, } +/// What the peer is to end up with for each prefix heard of: the route as +/// it goes out, or `None` for a prefix it is not to have. +type Verdicts = HashMap>>; + /// What an exporter knows of the learned routes it has passed on. struct Learned { reader: Reader, /// The verdicts so far on a table that is being given out anew, kept /// until the table is all there; and the leave to keep them. - gathered: Option>>>, + gathered: Option, permit: Option, /// The prefixes announced to the peer and not withdrawn since. announced: HashSet, @@ -299,6 +304,38 @@ impl Learned { self.announced.len() + self.stale.as_ref().map_or(0, HashSet::len) } + /// Takes the next changes of best route from the table: all of what + /// there is, up to a limit, so that it goes out together. Only what is + /// needed is taken while the table is held. + fn drain(&mut self) -> Vec { + let mut news = Vec::new(); + while news.len() < EXPORT_BATCH { + let progress = self.reader.poll(EXPORT_BATCH - news.len(), |event| { + news.push(match event { + Event::Reset => News::Reset, + Event::Synced => News::Synced, + Event::Best { + prefix, + peer, + attrs, + .. + } => News::Best { + prefix, + peer, + attrs: Arc::clone(attrs), + }, + Event::Withdrawn { prefix } => News::Withdrawn(prefix), + // Not given to a follower of the best routes. + Event::Routes { .. } => return, + }); + }); + if progress.caught_up || progress.events == 0 { + break; + } + } + news + } + fn new(reader: Reader) -> Self { Self { reader, @@ -500,46 +537,30 @@ impl Exporter { /// Takes the next changes of best route from the table and tells the /// peer of those that are for it. async fn follow(&mut self) -> Result<(), Finished> { - let towards = self.towards(); let Some(learned) = &mut self.learned else { return Ok(()); }; - // Only what is needed is taken while the table is held; and all - // of what there is, up to a limit, so that it goes out together. - let mut news = Vec::new(); - while news.len() < EXPORT_BATCH { - let progress = learned.reader.poll(EXPORT_BATCH - news.len(), |event| { - news.push(match event { - Event::Reset => News::Reset, - Event::Synced => News::Synced, - Event::Best { - prefix, - peer, - attrs, - .. - } => News::Best { - prefix, - peer, - attrs: Arc::clone(attrs), - }, - Event::Withdrawn { prefix } => News::Withdrawn(prefix), - // Not given to a follower of the best routes. - Event::Routes { .. } => return, - }); - }); - if progress.caught_up || progress.events == 0 { - break; - } + let news = learned.drain(); + match self.judge(news).await { + Some((verdicts, synced)) => self.flush(verdicts, synced).await, + // Not all there yet. + None => Ok(()), } + } + /// Decides what the peer is to end up with for each prefix the news + /// is about: the last word on a prefix counts. While the whole table + /// is being given out anew, which is shard by shard, the verdicts are + /// kept until it is all there, and this returns `None`: the routes + /// that share attributes are in every shard, and are only sent + /// together if gathered first. Otherwise the verdicts, and whether + /// the table has been given out whole. + async fn judge(&mut self, news: Vec) -> Option<(Verdicts, bool)> { + let towards = self.towards(); let session = &self.session; + let learned = self.learned.as_mut()?; let limits = session.export_limits.as_deref(); let egress = Self::egress(session, self.control, self.cluster_id, towards); - // What the peer is to end up with for each prefix heard of: the - // last word on it counts. While the whole table is being given out - // anew, which is shard by shard, this is kept until it is all - // there: the routes that share attributes are in every shard, and - // are only sent together if gathered first. let mut verdicts = learned.gathered.take().unwrap_or_default(); let mut synced = false; // Who the routes are from, asked of the table once for each. @@ -584,7 +605,7 @@ impl Exporter { // that, and not as someone else's route for it. Some(source) if session.families.carries(&prefix) - && !self.announced.contains(&prefix) + && !self.own_announced.contains(&prefix) && limits.is_none_or(|limits| limits.allows(&prefix)) => { learned.dressed(&attrs, source, prefix.address_family(), &egress) @@ -596,17 +617,25 @@ impl Exporter { } } } - if learned.permit.is_some() && !synced { - // Not all there yet. learned.gathered = Some(verdicts); - return Ok(()); + return None; } + Some((verdicts, synced)) + } - // Withdrawals first, then announcements, routes that go out alike - // together. + /// Sends the peer what the verdicts say: withdrawals first, then + /// announcements, routes that go out alike together. `synced` says + /// the table was given out whole, so what was not heard of again is + /// withdrawn too. + async fn flush(&mut self, verdicts: Verdicts, synced: bool) -> Result<(), Finished> { + let towards = self.towards(); + let session = &self.session; + let Some(learned) = &mut self.learned else { + return Ok(()); + }; let mut withdrawn: [Vec; 2] = Default::default(); - let mut announced: HashMap<(usize, IpVersion), (Arc, Vec)> = HashMap::new(); + let mut batches: HashMap<(usize, IpVersion), (Arc, Vec)> = HashMap::new(); for (prefix, verdict) in verdicts { let version = prefix.address_family(); // Heard of again: no longer in doubt, whichever way it goes. @@ -617,7 +646,7 @@ impl Exporter { match verdict { Some(dressed) => { learned.announced.insert(prefix); - announced + batches .entry((Arc::as_ptr(&dressed) as usize, version)) .or_insert_with(|| (dressed, Vec::new())) .1 @@ -632,15 +661,16 @@ impl Exporter { } if synced { learned.synced = true; - } - if synced && let Some(stale) = learned.stale.take() { - for prefix in stale { - withdrawn[version_index(prefix.address_family())].push(prefix.into()); + if let Some(stale) = learned.stale.take() { + for prefix in stale { + withdrawn[version_index(prefix.address_family())].push(prefix.into()); + } } } // What the peer would hold once this goes out: none of it goes if // that is more than it may. - let routes = self.announced.len() + learned.held_by_peer(); + let routes = self.own_announced.len() + learned.held_by_peer(); + let limits = session.export_limits.as_deref(); if let Some(limit) = limits.and_then(|limits| limits.exceeded_by(routes)) { return Err(over_limit(session, limit, routes)); } @@ -658,7 +688,7 @@ impl Exporter { session.send(Message::Update(update)).await?; } } - for ((_, version), (dressed, routes)) in announced { + for ((_, version), (dressed, routes)) in batches { ROUTES_EXPORTED .with_label_values(&["announced"]) .inc_by(routes.len() as u64); @@ -685,7 +715,7 @@ impl Exporter { async fn send_all(&mut self) -> Result<(), Finished> { let routes = self.local_rib.snapshot().await.map_err(|_| Finished)?; let current: HashSet = routes.iter().map(|route| route.prefix).collect(); - let gone: Vec = self.announced.difference(¤t).copied().collect(); + let gone: Vec = self.own_announced.difference(¤t).copied().collect(); for prefix in gone { self.withdraw(prefix).await?; } @@ -715,8 +745,8 @@ impl Exporter { return Ok(()); } // What the peer would hold with this route. - let routes = self.announced.len() - + usize::from(!self.announced.contains(&route.prefix)) + let routes = self.own_announced.len() + + usize::from(!self.own_announced.contains(&route.prefix)) + self.learned.as_ref().map_or(0, Learned::held_by_peer); if let Some(limit) = limits.and_then(|limits| limits.exceeded_by(routes)) { return Err(over_limit(session, limit, routes)); @@ -744,12 +774,12 @@ impl Exporter { export::finish(&mut update.path_attributes, &towards); debug!(peer = %session.peer, prefix = %route.prefix, "Announcing local route"); session.send(Message::Update(update)).await?; - self.announced.insert(route.prefix); + self.own_announced.insert(route.prefix); Ok(()) } async fn withdraw(&mut self, prefix: IpPrefix) -> Result<(), Finished> { - if !self.announced.remove(&prefix) { + if !self.own_announced.remove(&prefix) { // The peer was never told of it. return Ok(()); } diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index 35e36ff..c52b3b4 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -1,4 +1,4 @@ -use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::str::FromStr; use std::sync::Arc; use std::time::Duration; @@ -132,129 +132,39 @@ impl Config { .ok_or_else(|| ConfigError::ClusterId(proto.cluster_id.clone()))? }; - let listen_endpoints = if proto.listen_endpoints.is_empty() { - vec![ - SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), DEFAULT_BGP_PORT), - SocketAddr::new( - IpAddr::V6(std::net::Ipv6Addr::UNSPECIFIED), - DEFAULT_BGP_PORT, - ), - ] - } else { - proto - .listen_endpoints - .iter() - .map(|ep| parse_endpoint(&ep.address, ep.port)) - .collect::>()? - }; + let listen_endpoints = parse_listen_endpoints(&proto.listen_endpoints)?; - let export = Arc::new(match proto.export.as_ref() { - Some(export) => parse_export(export)?, - None => ExportPolicy::default(), - }); - let export_limits = match proto.export_limits.as_ref() { - Some(limits) => parse_export_limits(limits) - .map_err(ConfigError::ExportLimits)? - .map(Arc::new), - None => None, + let defaults = PeerDefaults { + export: Arc::new(match proto.export.as_ref() { + Some(export) => parse_export(export)?, + None => ExportPolicy::default(), + }), + export_limits: match proto.export_limits.as_ref() { + Some(limits) => parse_export_limits(limits) + .map_err(ConfigError::ExportLimits)? + .map(Arc::new), + None => None, + }, + graceful_restart: match proto.graceful_restart.as_ref() { + Some(own) => GracefulRestartConfig::default().overridden_by(own)?, + None => GracefulRestartConfig::default(), + }, }; - let graceful_restart = match proto.graceful_restart.as_ref() { - Some(own) => GracefulRestartConfig::default().overridden_by(own)?, - None => GracefulRestartConfig::default(), + let mut peers = parse_peers(proto.peers, proto.local_asn, &defaults)?; + check_roles( + &peers, + &defaults.export, + proto.accept_unconfigured_peers, + proto.local_asn, + )?; + let originate = parse_originate(&proto.originate)?; + let ours = Ours { + local_asn: proto.local_asn, + router_id, + cluster_id, + control_communities: proto.control_communities, }; - 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, &export, export_limits.as_ref(), graceful_restart) - .map_err(rejected)?; - // RFC 9234 is about what networks are to each other. - if peer.role.is_some() && peer.remote_asn == proto.local_asn { - return Err(rejected(PeerProblem::RoleInOwnAs)); - } - if peer.reflector_client && peer.remote_asn != proto.local_asn { - return Err(rejected(PeerProblem::ReflectorClientOutsideAs)); - } - // 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 - && earlier.endpoint.interface == peer.endpoint.interface - }) { - return Err(rejected(PeerProblem::DuplicateAddress)); - } - peers.push(peer); - } - // Where routes learned from one peer go on to others, which go - // where is decided by the peers' roles: a peer in another AS - // without one would have its routes go to everyone, and be told - // of everything. - let learned_go_on = - export.learned_routes || peers.iter().any(|peer| peer.export.learned_routes); - if learned_go_on - && let Some(peer) = peers - .iter() - .find(|peer| peer.remote_asn != proto.local_asn && peer.role.is_none()) - { - return Err(ConfigError::Peer { - peer: peer.name.clone(), - problem: PeerProblem::NoRole, - }); - } - if proto.accept_unconfigured_peers && export.learned_routes { - return Err(ConfigError::UnconfiguredPeersWithoutRole); - } - - let mut originate: Vec = Vec::with_capacity(proto.originate.len()); - for entry in &proto.originate { - let rejected = |problem| ConfigError::Originate { - prefix: entry.prefix.clone(), - problem, - }; - let route = parse_originated_route(entry).map_err(rejected)?; - if originate - .iter() - .any(|earlier| earlier.prefix == route.prefix) - { - return Err(rejected(OriginateProblem::Duplicate)); - } - originate.push(route); - } - // What we originate is not taken from other networks. - let mut own_prefixes = PrefixSet::default(); - let mut originated = PrefixSet::default(); - for route in &originate { - own_prefixes.allow(route.prefix, route.prefix.address_family().bits()); - originated.allow(route.prefix, route.prefix.length()); - } - let own_prefixes = Arc::new(own_prefixes); - let originated = Arc::new(originated); - for peer in &mut peers { - if peer.remote_asn == proto.local_asn { - peer.import.originated = Arc::clone(&originated); - peer.import.reflection = Some(Reflection { - router_id, - cluster_id, - }); - } else { - peer.import.own_prefixes = Arc::clone(&own_prefixes); - // What is ours to say is not taken from other networks, - // unless they are to say it: customers that steer their - // own routes. - if !peer.import.accept_control_communities { - peer.import.scrub = Some(Scrub { - local_asn: proto.local_asn, - zero: proto.control_communities, - }); - } - } - } + ours.tell_peers(&mut peers, &originate); let http_endpoint = match proto.http_endpoint.into_option() { Some(ep) if ep.port == 0 => return Err(ConfigError::HttpPort), @@ -287,16 +197,167 @@ impl Config { http_endpoint, kernel: proto.kernel.as_ref().map(parse_kernel), rpki, - export, - export_limits, + export: defaults.export, + export_limits: defaults.export_limits, control: proto.control_communities.then_some(Control { local_asn: proto.local_asn, }), - graceful_restart, + graceful_restart: defaults.graceful_restart, }) } } +/// Where BGP connections are taken: as configured, or on every address. +fn parse_listen_endpoints( + endpoints: &[proto::TcpEndpoint], +) -> Result, ConfigError> { + if endpoints.is_empty() { + return Ok(vec![ + SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), DEFAULT_BGP_PORT), + SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), DEFAULT_BGP_PORT), + ]); + } + endpoints + .iter() + .map(|ep| parse_endpoint(&ep.address, ep.port)) + .collect() +} + +/// What a peer gets that has no setting of its own. +struct PeerDefaults { + export: Arc, + export_limits: Option>, + graceful_restart: GracefulRestartConfig, +} + +/// The peers, each in order on its own and told from the others: by name, +/// which the dialer reports, and by address, which a peer that connects +/// to us is known by. +fn parse_peers( + peers: Vec, + local_asn: u32, + defaults: &PeerDefaults, +) -> Result, ConfigError> { + let mut parsed: Vec = Vec::with_capacity(peers.len()); + for peer in peers { + let name = peer.name.clone(); + let rejected = |problem| ConfigError::Peer { + peer: name.clone(), + problem, + }; + let peer = parse_peer(peer, defaults).map_err(rejected)?; + // RFC 9234 is about what networks are to each other. + if peer.role.is_some() && peer.remote_asn == local_asn { + return Err(rejected(PeerProblem::RoleInOwnAs)); + } + if peer.reflector_client && peer.remote_asn != local_asn { + return Err(rejected(PeerProblem::ReflectorClientOutsideAs)); + } + if parsed.iter().any(|earlier| earlier.name == peer.name) { + return Err(rejected(PeerProblem::DuplicateName)); + } + if parsed.iter().any(|earlier| { + earlier.endpoint.ip == peer.endpoint.ip + && earlier.endpoint.interface == peer.endpoint.interface + }) { + return Err(rejected(PeerProblem::DuplicateAddress)); + } + parsed.push(peer); + } + Ok(parsed) +} + +/// Where routes learned from one peer go on to others, which go where is +/// decided by the peers' roles: a peer in another AS without one would +/// have its routes go to everyone, and be told of everything. Peers that +/// are not configured have no role either. +fn check_roles( + peers: &[PeerConfig], + export: &ExportPolicy, + accept_unconfigured_peers: bool, + local_asn: u32, +) -> Result<(), ConfigError> { + let learned_go_on = + export.learned_routes || peers.iter().any(|peer| peer.export.learned_routes); + if learned_go_on + && let Some(peer) = peers + .iter() + .find(|peer| peer.remote_asn != local_asn && peer.role.is_none()) + { + return Err(ConfigError::Peer { + peer: peer.name.clone(), + problem: PeerProblem::NoRole, + }); + } + if accept_unconfigured_peers && export.learned_routes { + return Err(ConfigError::UnconfiguredPeersWithoutRole); + } + Ok(()) +} + +/// The routes this router originates, each prefix once. +fn parse_originate(entries: &[proto::OriginatedRoute]) -> Result, ConfigError> { + let mut originate: Vec = Vec::with_capacity(entries.len()); + for entry in entries { + let rejected = |problem| ConfigError::Originate { + prefix: entry.prefix.clone(), + problem, + }; + let route = parse_originated_route(entry).map_err(rejected)?; + if originate + .iter() + .any(|earlier| earlier.prefix == route.prefix) + { + return Err(rejected(OriginateProblem::Duplicate)); + } + originate.push(route); + } + Ok(originate) +} + +/// What is this router's own, which every peer's import policy has to +/// know of. +struct Ours { + local_asn: u32, + router_id: Ipv4Addr, + cluster_id: Ipv4Addr, + control_communities: bool, +} + +impl Ours { + /// Gives each peer's import policy what of ours it is to look out + /// for: our routers, what we originate and how we would reflect; the + /// others, our address space and our communities, which are not + /// theirs to send unless they are to steer their own routes. + fn tell_peers(&self, peers: &mut [PeerConfig], originate: &[LocalRoute]) { + let mut own_prefixes = PrefixSet::default(); + let mut originated = PrefixSet::default(); + for route in originate { + own_prefixes.allow(route.prefix, route.prefix.address_family().bits()); + originated.allow(route.prefix, route.prefix.length()); + } + let own_prefixes = Arc::new(own_prefixes); + let originated = Arc::new(originated); + for peer in peers { + if peer.remote_asn == self.local_asn { + peer.import.originated = Arc::clone(&originated); + peer.import.reflection = Some(Reflection { + router_id: self.router_id, + cluster_id: self.cluster_id, + }); + } else { + peer.import.own_prefixes = Arc::clone(&own_prefixes); + if !peer.import.accept_control_communities { + peer.import.scrub = Some(Scrub { + local_asn: self.local_asn, + zero: self.control_communities, + }); + } + } + } + } +} + /// A BGP identifier (RFC 4271 §4.2) as the configuration writes one: a /// dotted quad other than 0.0.0.0, which peers refuse, as we do theirs. fn parse_identifier(text: &str) -> Option { @@ -360,12 +421,7 @@ fn parse_export_limits( Ok((limits != ExportLimits::default()).then_some(limits)) } -fn parse_peer( - peer: Peer, - export: &Arc, - export_limits: Option<&Arc>, - graceful_restart: GracefulRestartConfig, -) -> Result { +fn parse_peer(peer: Peer, defaults: &PeerDefaults) -> Result { let ep = peer.endpoint.as_ref().ok_or(PeerProblem::MissingEndpoint)?; let endpoint = parse_port(ep.port) .and_then(|port| Ok(Endpoint::parse(&ep.address, port)?)) @@ -455,17 +511,17 @@ fn parse_peer( require_role: peer.require_role, export: match peer.export.as_ref() { Some(own) => Arc::new(parse_export(own)?), - None => Arc::clone(export), + None => Arc::clone(&defaults.export), }, export_limits: match peer.export_limits.as_ref() { Some(own) => parse_export_limits(own) .map_err(PeerProblem::ExportLimits)? .map(Arc::new), - None => export_limits.cloned(), + None => defaults.export_limits.clone(), }, graceful_restart: match peer.graceful_restart.as_ref() { - Some(own) => graceful_restart.overridden_by(own)?, - None => graceful_restart, + Some(own) => defaults.graceful_restart.overridden_by(own)?, + None => defaults.graceful_restart, }, groups: peer.groups, reflector_client: peer.route_reflector_client, diff --git a/crates/server/src/connection.rs b/crates/server/src/connection.rs index eb53589..33a35e9 100644 --- a/crates/server/src/connection.rs +++ b/crates/server/src/connection.rs @@ -41,14 +41,13 @@ use crate::config::PeerConfig; use crate::error::SessionError; use crate::family::FamilySet; use crate::fsm::{Action, Event, Fsm, FsmConfig, Negotiated, Received}; +use crate::holddown::{HeldDown, HoldDowns}; use crate::metrics::{ EXPORT_LIMIT_EXCEEDED, MESSAGES_RECEIVED, MESSAGES_SENT, PEER_ROUTES, PREFIX_LIMIT_EXCEEDED, SESSIONS_ESTABLISHED_TOTAL, UPDATE_ERRORS, message_type_name, }; use crate::panic::contained; -use crate::policy::{ - self, ExportLimits, ExportPolicy, HeldDown, HoldDowns, ImportPolicy, PeerRole, -}; +use crate::policy::{self, ExportLimits, ExportPolicy, ImportPolicy, PeerRole}; use crate::reach::{Reach, neighbor_at}; use crate::refusals::{PeerRefusals, RefusalLog}; use crate::restart::{self, Agreed, GracefulRestartConfig, Leave, Restarts}; @@ -226,11 +225,32 @@ pub async fn run_session( Err(panic) => Err(session.close_after_panic(panic).await), }; let established = session.established.is_some(); + hold_down_after(&session, &result); + // Whatever ended the session, routes learned over it are gone + // (RFC 4271 §8.2.2), unless the peer is restarting (RFC 4724 §4.2). + // Done before the registration drops, so that the routes are + // withdrawn, or marked stale, before the session is seen to be down. + if let Some(established) = session.established.take() { + release_routes(&session, established, &result).await; + } + session.close().await; + SessionEnd { + established, + result, + } +} + +/// Keeps the peer down for a while if the session that just ended says +/// to: a peer that sent far more than it should, or that we were to send +/// far more than we should, will be so again on the next session, and is +/// left alone until it or we are put right. A configured peer whose +/// sessions keep coming up and failing is slowed down; not one that is +/// nobody in particular, such as are only taken on a test network, where +/// many share an address. +fn hold_down_after(session: &Session, result: &Result<(), SessionError>) { + let address = PeerAddress::from(session.peer_addr); let hold_downs = &session.server.hold_downs; - // A peer that sent far more than it should, or that we were to send - // far more than we should, will be so again on the next session; it - // is left alone for a while, until it or we are put right. - let over_limit = match &result { + let over_limit = match result { Err(SessionError::PrefixLimitExceeded { .. }) => { Some((HeldDown::OverLimit, &*PREFIX_LIMIT_EXCEEDED)) } @@ -240,89 +260,90 @@ pub async fn run_session( _ => None, }; if let Some((why, exceeded)) = over_limit { - hold_downs.hold(peer_addr.into(), session.import.hold_down, why); + hold_downs.hold(address, session.peer.import.hold_down, why); exceeded .with_label_values(&[session.refusals.label()]) .inc(); } - // A configured peer whose sessions keep coming up and failing is - // slowed down. Not one that is nobody in particular: such peers are - // only taken on a test network, where many share an address. - if let (Some(up), Some(_)) = (&session.established, &session.peer_name) - && let Some(hold) = - hold_downs.session_ended(peer_addr.into(), up.since.elapsed(), result.is_err()) + if let (Some(up), Some(_)) = (&session.established, &session.peer.name) + && let Some(hold) = hold_downs.session_ended(address, up.since.elapsed(), result.is_err()) { warn!( - peer_addr = %socket_text(peer_addr), + peer_addr = %socket_text(session.peer_addr), hold_down = ?hold, "The peer's sessions keep failing soon after they come up; keeping it down for a while" ); } +} - // Whatever ended the session, routes learned over it are gone - // (RFC 4271 §8.2.2), unless the peer is restarting (RFC 4724 §4.2). - // Do this before the registration drops so that the routes are - // withdrawn, or marked stale, before the session is seen to be down. - if let Some(established) = session.established.take() { - let leaving = *session.leave.borrow(); - let keep_for = established - .graceful_restart - .as_ref() - .filter(|agreed| leaving.is_none() && restart::ends_in_restart(&result, agreed)) - .map(|agreed| { - let theirs = Duration::from_secs(agreed.restart_time.into()); - ( - restart::versions(&agreed.families), - theirs.min(session.graceful_restart.stale_routes), - ) - }) - .filter(|(versions, time)| !versions.is_empty() && !time.is_zero()); - let Established { - peer_id, - rib_peer, - routes, - _registration, - .. - } = established; - // The peer has no session to count its routes for. - // Fails only if the session never set it, which is nothing to do. - let _ = PEER_ROUTES.remove_label_values(&[session.refusals.label()]); - if leaving == Some(Leave::Restart) { - // The daemon is about to go, and its routes with it; the - // kernel keeps its own. Taking them out first would only hold - // the restart up. - routes.forget(); - } else if let Some((versions, time)) = keep_for { - let kept = routes.keep(versions.clone()).await; - session.server.restarts.keep( - (peer_addr.into(), peer_id.asn), - peer_addr, - session.peer_name.clone(), - rib_peer, - versions, - kept, - time, - ); - info!( - peer = %peer_id, - kept_routes = kept, - kept_for = ?time, - "Session ended; the peer is restarting, and its routes are kept meanwhile" - ); - } else { - let outcome = routes.remove().await; - info!( - peer = %peer_id, - removed_routes = outcome.removed, - best_changed = outcome.best_changed, - "Session ended" - ); - } +/// Whether, and for how long, the routes of a session that ended as +/// `result` says are to be kept for the peer's next session (RFC 4724 +/// §4.2): the IP versions whose routes are, and the time. `None` if they +/// are to go now. +fn keep_for( + session: &Session, + established: &Established, + result: &Result<(), SessionError>, +) -> Option<(Vec, Duration)> { + let agreed = established.graceful_restart.as_ref()?; + if session.leave.borrow().is_some() || !restart::ends_in_restart(result, agreed) { + return None; } - session.close().await; - SessionEnd { - established, - result, + let versions = restart::versions(&agreed.families); + let theirs = Duration::from_secs(agreed.restart_time.into()); + let time = theirs.min(session.peer.graceful_restart.stale_routes); + (!versions.is_empty() && !time.is_zero()).then_some((versions, time)) +} + +/// What becomes of the routes of a session that has ended: kept for the +/// peer's next session if it is restarting, left where they are if the +/// daemon is, and removed otherwise. +async fn release_routes( + session: &Session, + established: Established, + result: &Result<(), SessionError>, +) { + let keep = keep_for(session, &established, result); + let Established { + peer_id, + rib_peer, + routes, + _registration, + .. + } = established; + // The peer has no session to count its routes for. + // Fails only if the session never set it, which is nothing to do. + let _ = PEER_ROUTES.remove_label_values(&[session.refusals.label()]); + if *session.leave.borrow() == Some(Leave::Restart) { + // The daemon is about to go, and its routes with it; the kernel + // keeps its own. Taking them out first would only hold the + // restart up. + routes.forget(); + } else if let Some((versions, time)) = keep { + let kept = routes.keep(versions.clone()).await; + session.server.restarts.keep( + (session.peer_addr.into(), peer_id.asn), + session.peer_addr, + session.peer.name.clone(), + rib_peer, + versions, + kept, + time, + ); + info!( + peer = %peer_id, + kept_routes = kept, + kept_for = ?time, + "Session ended; the peer is restarting, and its routes are kept meanwhile" + ); + } else { + let outcome = routes.remove().await; + info!( + peer = %peer_id, + removed_routes = outcome.removed, + best_changed = outcome.best_changed, + "Session ended" + ); } } @@ -364,14 +385,25 @@ struct Established { /// again yet, if there are any. stale: Option, /// Where the sweeps of stale routes say what they did. - swept: ( - mpsc::UnboundedSender, - mpsc::UnboundedReceiver, - ), + swept: Sweeps, /// Keeps the session listed in the registry. _registration: SessionGuard, } +/// Where the sweeps of a session's stale routes, which run on their own +/// threads, say what they did. +struct Sweeps { + report: mpsc::UnboundedSender, + reported: mpsc::UnboundedReceiver, +} + +impl Default for Sweeps { + fn default() -> Self { + let (report, reported) = mpsc::unbounded_channel(); + Self { report, reported } + } +} + /// The routes kept from a restarting peer's last session, while its new /// one sends them again: those of the IP versions it has not sent an /// End-of-RIB marker for yet, which are kept until `until` at most. @@ -617,24 +649,10 @@ struct Session { local_addr: SocketAddr, direction: Direction, server: Arc, - /// Configured name of the peer, if it is a configured one. - peer_name: Option, - /// Which of the peer's routes are taken. - import: Arc, - /// What the peer is to us, if that is known. - role: Option, - /// What the peer is told, and how. - export: Arc, - /// Limits on what the peer is told, if it is one they are for. - export_limits: Option>, - /// The groups the session is in, for control communities. - groups: Arc<[u32]>, - /// Whether the peer is a route reflector client of ours (RFC 4456). - reflector_client: bool, + /// What is configured for the peer. + peer: PeerSettings, /// Where what is refused is noted. refusals: Arc, - /// Graceful restart with the peer, as configured. - graceful_restart: GracefulRestartConfig, /// Says when the daemon is leaving, and how. leave: watch::Receiver>, } @@ -656,7 +674,7 @@ impl Session { local_asn: server.local_asn, hold_time_secs: peer.hold_time_secs, expected_asn: peer.expected_asn, - families: peer.families, + families: peer.families.clone(), add_path: peer.add_path, role: peer.role, require_role: peer.require_role, @@ -689,15 +707,8 @@ impl Session { local_addr, direction, server, - peer_name: peer.name, - import: peer.import, - role: peer.role, - export: peer.export, - export_limits: peer.export_limits, - groups: peer.groups, - reflector_client: peer.reflector_client, + peer, said: Said::new(), - graceful_restart: peer.graceful_restart, leave, }; Ok((session, first_actions)) @@ -883,7 +894,7 @@ impl Session { SessionError::PrefixLimitExceeded { .. } => warn!( peer_addr = %socket_text(self.peer_addr), err = %e, - hold_down = ?self.import.hold_down, + hold_down = ?self.peer.import.hold_down, "Ending the session and keeping the peer down" ), _ => error!( @@ -934,7 +945,7 @@ impl Session { self.peer_addr.into(), negotiated.peer_identifier, self.direction, - self.peer_name.is_some(), + self.peer.name.is_some(), ); if self.collision.is_none() { info!(peer_addr = %socket_text(self.peer_addr), "Closing connection: lost collision resolution"); @@ -968,7 +979,7 @@ impl Session { interface: PeerAddress::from(self.peer_addr).interface, ibgp: negotiated.ibgp, // The configuration allows it only inside our AS. - reflector_client: self.reflector_client && negotiated.ibgp, + reflector_client: self.peer.reflector_client && negotiated.ibgp, }; // A peer back from a restart takes over the routes kept of it. let kept = self @@ -987,50 +998,9 @@ impl Session { collision.set_established(); } let (outbound_tx, outbound) = mpsc::channel(OUTBOUND_QUEUE_CAPACITY); - let mut handle = SessionHandle::new( - peer_id, - self.peer_name.clone(), - self.direction, - self.local_addr, - negotiated, - rib_peer, - outbound_tx, - ); - handle.role = self.role; - handle.role_confirmed = negotiated.role_confirmed; - handle.export = Arc::clone(&self.export); - // To a network above or beside us, or one we know nothing of. - handle.export_limits = self - .export_limits - .clone() - .filter(|_| ExportLimits::apply_to(self.role, negotiated.ibgp)); + let mut handle = self.session_handle(negotiated, peer_id, rib_peer, outbound_tx); let ending = handle.listen_for_end(); - handle.groups = Arc::clone(&self.groups); - handle.reflector_client = self.reflector_client && negotiated.ibgp; - // Routes from other networks are validated, where there is data - // to validate against and the peer's policy does not say to take - // invalid ones anyway. - let subject = Subject::of( - &self.import, - self.role, - negotiated.peer_asn, - negotiated.ibgp, - ); - // And where routes go into the kernel, a route has to have a next - // hop the kernel can use. - let reach = self - .server - .interfaces - .clone() - .map(|interfaces| Reach::new(interfaces, neighbor_at(self.peer_addr))); - let subject = subject.filter(|_| self.server.rpki.is_some()); - let validation = - Validator::new(self.server.rpki.clone().zip(subject), reach).map(|validator| { - Validation { - validator, - changes: handle.validate(subject), - } - }); + let validation = self.validation(&mut handle, negotiated); self.established = Some(Established { since: Instant::now(), validation, @@ -1040,22 +1010,13 @@ impl Session { graceful_restart: negotiated.graceful_restart.clone(), end_of_rib: FamilySet::default(), stale: None, - swept: mpsc::unbounded_channel(), + swept: Sweeps::default(), peer_id, rib_peer, outbound, ending, routes: PeerRoutes::new(Arc::clone(&self.server.rib), rib_peer), - from: policy::RouteSource { - peer_asn: negotiated.peer_asn, - local_asn: self.server.local_asn, - ibgp: negotiated.ibgp, - four_octet_asn: negotiated.four_octet_asn, - role: self.role, - peer_address: self.peer_addr.ip(), - local_address: self.local_addr.ip(), - extended_next_hop: negotiated.extended_next_hop, - }, + from: self.route_source(negotiated), counters: Arc::clone(&handle.counters), routes_metric: PEER_ROUTES.with_label_values(&[self.refusals.label()]), _registration: self.server.registry.register(handle), @@ -1066,6 +1027,80 @@ impl Session { Ok(()) } + /// The session as the rest of the server is to see it: what was + /// negotiated, and what is configured for the peer. + fn session_handle( + &self, + negotiated: &Negotiated, + peer_id: PeerId, + rib_peer: PeerIndex, + outbound: mpsc::Sender, + ) -> SessionHandle { + let peer = &self.peer; + let mut handle = SessionHandle::new( + peer_id, + peer.name.clone(), + self.direction, + self.local_addr, + negotiated, + rib_peer, + outbound, + ); + handle.role = peer.role; + handle.role_confirmed = negotiated.role_confirmed; + handle.export = Arc::clone(&peer.export); + // To a network above or beside us, or one we know nothing of. + handle.export_limits = peer + .export_limits + .clone() + .filter(|_| ExportLimits::apply_to(peer.role, negotiated.ibgp)); + handle.groups = Arc::clone(&peer.groups); + // The configuration allows it only inside our AS. + handle.reflector_client = peer.reflector_client && negotiated.ibgp; + handle + } + + /// How the peer's routes are checked as the world changes, if they + /// are: against the RPKI, where there is data to validate against and + /// the peer's policy does not say to take invalid routes anyway; and, + /// where routes go into the kernel, for a next hop the kernel can use. + fn validation( + &self, + handle: &mut SessionHandle, + negotiated: &Negotiated, + ) -> Option { + let subject = Subject::of( + &self.peer.import, + self.peer.role, + negotiated.peer_asn, + negotiated.ibgp, + ) + .filter(|_| self.server.rpki.is_some()); + let reach = self + .server + .interfaces + .clone() + .map(|interfaces| Reach::new(interfaces, neighbor_at(self.peer_addr))); + Validator::new(self.server.rpki.clone().zip(subject), reach).map(|validator| Validation { + validator, + changes: handle.validate(subject), + }) + } + + /// Whose routes the import policy is judging. + fn route_source(&self, negotiated: &Negotiated) -> policy::RouteSource { + policy::RouteSource { + peer_asn: negotiated.peer_asn, + local_asn: self.server.local_asn, + ibgp: negotiated.ibgp, + four_octet_asn: negotiated.four_octet_asn, + role: self.peer.role, + peer_address: self.peer_addr.ip(), + local_address: self.local_addr.ip(), + extended_next_hop: negotiated.extended_next_hop, + } + } + /// Takes over the routes kept from the peer's last session, which are /// stale until it sends them again (RFC 4724 §4.2). Those of the /// families whose forwarding it did not keep over its restart, or all @@ -1099,13 +1134,13 @@ impl Session { &self.server.rib, established.rib_peer, go, - &established.swept.0, + &established.swept.report, ); } if !stay.is_empty() { established.stale = Some(Stale { versions: stay, - until: Box::pin(sleep(self.graceful_restart.stale_routes)), + until: Box::pin(sleep(self.peer.graceful_restart.stale_routes)), }); } } @@ -1130,10 +1165,10 @@ impl Session { &self.server.rib, established.rib_peer, vec![version], - &established.swept.0, + &established.swept.report, ); } - if let Some(name) = &self.peer_name + if let Some(name) = &self.peer.name && established .families .iter() @@ -1158,7 +1193,7 @@ impl Session { &self.server.rib, established.rib_peer, stale.versions, - &established.swept.0, + &established.swept.report, ); } } @@ -1178,7 +1213,7 @@ impl Session { } // Every route swept away was stale. established.counters.clear_stale(outcome.removed); - match established.account(outcome, 0, &self.import) { + match established.account(outcome, 0, &self.peer.import) { Ok(()) => vec![], Err(e) => self.fsm.handle(Event::LocalFailure(e)), } @@ -1193,7 +1228,7 @@ impl Session { let Some(established) = &mut self.established else { return Ok(()); }; - let (import, held) = established.import(&self.import, &self.refusals); + let (import, held) = established.import(&self.peer.import, &self.refusals); let routes::Applied { outcome, rejected } = routes::apply(&self.server.rib, &import, held, update, attributes_wire)?; // One line for every UPDATE is too much to log by default: a full @@ -1207,7 +1242,7 @@ impl Session { best_changed = outcome.best_changed, "Received UPDATE" ); - established.account(&outcome, rejected, &self.import) + established.account(&outcome, rejected, &self.peer.import) } /// Looks at the peer's routes again after the validation data has @@ -1217,7 +1252,7 @@ impl Session { let Some(established) = &mut self.established else { return vec![]; }; - let (import, held) = established.import(&self.import, &self.refusals); + let (import, held) = established.import(&self.peer.import, &self.refusals); let routes::Applied { outcome, rejected } = routes::revalidate(&self.server.rib, &import, held, suspects); if outcome != Outcome::default() { @@ -1229,7 +1264,7 @@ impl Session { "Revalidated the peer's routes" ); } - if let Err(e) = established.account(&outcome, rejected, &self.import) { + if let Err(e) = established.account(&outcome, rejected, &self.peer.import) { return self.fsm.handle(Event::LocalFailure(e)); } // Invalid routes that there was no room to hold may be acceptable @@ -1343,7 +1378,7 @@ async fn from_server(established: &mut Option) -> Wake { Some(error) = ending.recv() => Wake::Event(Event::LocalFailure(error)), Some(msg) = outbound.recv() => Wake::Event(Event::Outbound(msg)), Some(suspects) = invalidated => Wake::Revalidate(suspects), - Some(outcome) = swept.1.recv() => Wake::Swept(outcome), + Some(outcome) = swept.reported.recv() => Wake::Swept(outcome), () = stale_timeout => Wake::StaleTimeout, // The registry holds the sending halves for as long as the session // is registered, so neither closes under us; if they somehow did, diff --git a/crates/server/src/export.rs b/crates/server/src/export.rs index 52083bf..8acc6e3 100644 --- a/crates/server/src/export.rs +++ b/crates/server/src/export.rs @@ -161,26 +161,46 @@ impl Egress<'_> { source: Source, routes: IpVersion, ) -> Option { + let reflected = source.ibgp && self.towards.ibgp; + if !self.may_go(&attributes, source, reflected) { + return None; + } + let came_with = next_hop.or_else(|| { + attributes.iter().find_map(|attribute| match attribute { + PathAttribute::NextHop(next_hop) => Some(NlriNextHop::Ipv4(next_hop.0)), + _ => None, + }) + }); + if !self.rewrite(&mut attributes, source, reflected) { + return None; + } + let next_hop = self.next_hop_for(routes, came_with)?; + Some(Outgoing { + attributes, + next_hop, + }) + } + + /// Whether a route with these attributes, from `source`, is for this + /// session at all, as far as who may be told of what goes: the export + /// policy, route reflection, the AS path, the well-known communities + /// and the Only to Customer mark. + fn may_go(&self, attributes: &[PathAttribute], source: Source, reflected: bool) -> bool { let towards = &self.towards; if !self.policy.learned_routes || source.peer == self.rib_peer { - return None; + return false; } // From one router of our AS to another only by way of a route // reflector, which we are only for our clients: what a client sends // goes to all our routers, and what the others send goes to the // clients (RFC 4456 §6). - let reflected = source.ibgp && towards.ibgp; if reflected && !source.reflector_client && !self.reflection.client { - return None; + return false; } - let mut came_with = next_hop; let mut marked = false; let mut on_path = false; - for attribute in &attributes { + for attribute in attributes { match attribute { - PathAttribute::NextHop(next_hop) => { - came_with = came_with.or(Some(NlriNextHop::Ipv4(next_hop.0))); - } PathAttribute::OnlyToCustomer(_) => marked = true, PathAttribute::AsPath(path) => { on_path = path @@ -193,24 +213,36 @@ impl Egress<'_> { } // The peer would refuse it as having been there before. if on_path && !towards.ibgp { - return None; + return false; } // Whoever sent it said how far it may go (RFC 1997). On a route // of our own the same communities are for those who get it from // us, and it goes out with them. - if !well_known_allow(communities_of(&attributes).0, towards.ibgp) { - return None; + if !well_known_allow(communities_of(attributes).0, towards.ibgp) { + return false; } // It came down or across once: it goes on down only. - if marked + let only_down = marked && !towards.ibgp && towards .role - .is_some_and(|role| !role.may_be_sent_only_to_customer_routes()) - { - return None; - } + .is_some_and(|role| !role.may_be_sent_only_to_customer_routes()); + !only_down + } + /// Makes the attributes those of the route as this session announces + /// it: what is ours to say is taken off, what is for routers of the + /// AS it was learned in stays only inside ours, a reflected route gets + /// its reflection attributes, and the path and LOCAL_PREF are as the + /// peer is to see them. Then the session's own marks ([`Egress::dress`]). + /// `false` if the route is not to go after all. + fn rewrite( + &self, + attributes: &mut Vec, + source: Source, + reflected: bool, + ) -> bool { + let towards = &self.towards; let transparent = self.transparent(); attributes.retain_mut(|attribute| match attribute { // Ours to say, each way. @@ -237,8 +269,8 @@ impl Egress<'_> { } _ => true, }); - if reflected && !self.reflect(&mut attributes, source) { - return None; + if reflected && !self.reflect(attributes, source) { + return false; } if towards.ibgp { if !attributes @@ -246,7 +278,7 @@ impl Egress<'_> { .any(|attribute| matches!(attribute, PathAttribute::LocalPref(_))) { marks::put( - &mut attributes, + attributes, PathAttribute::LocalPref(LocalPrefPathAttribute(DEFAULT_LOCAL_PREF)), ); } @@ -258,17 +290,24 @@ impl Egress<'_> { }) { Some(path) => prepend(path, ours), None => marks::put( - &mut attributes, + attributes, AsPathAttribute::from_asns(vec![towards.local_asn]), ), } } - if !self.dress(&mut attributes) { - return None; - } + self.dress(attributes) + } - // Ourselves, unless we are a route server and the client can use - // the next hop the route came with. + /// The next hop of a route for prefixes of IP version `routes`: + /// ourselves, unless we are a route server and the client can use the + /// next hop the route came with. `None` where our address on the + /// session is of the wrong kind for the routes. + fn next_hop_for( + &self, + routes: IpVersion, + came_with: Option, + ) -> Option { + let towards = &self.towards; let ours = match (routes, towards.next_hop) { (IpVersion::Ipv4, IpAddr::V4(ours)) => NlriNextHop::Ipv4(ours), (IpVersion::Ipv6, IpAddr::V6(ours)) => ipv6_next_hop(ours), @@ -278,13 +317,11 @@ impl Egress<'_> { let same_kind = |theirs: &NlriNextHop| { matches!(theirs, NlriNextHop::Ipv4(_)) == matches!(ours, NlriNextHop::Ipv4(_)) }; - let next_hop = came_with - .filter(|theirs| transparent && same_kind(theirs)) - .unwrap_or(ours); - Some(Outgoing { - attributes, - next_hop, - }) + Some( + came_with + .filter(|theirs| self.transparent() && same_kind(theirs)) + .unwrap_or(ours), + ) } } diff --git a/crates/server/src/fsm.rs b/crates/server/src/fsm.rs index 0902c68..e803c38 100644 --- a/crates/server/src/fsm.rs +++ b/crates/server/src/fsm.rs @@ -352,34 +352,9 @@ impl Fsm { }); } - // RFC 9234 §4.2: the two ends have to agree on what they are to - // each other. - let mut their_roles = open - .capabilities() - .filter_map(|capability| match capability { - Capability::BgpRole { role } => Some(*role), - _ => None, - }); - let their_role = their_roles.next(); - let role_mismatch = |theirs| SessionError::RoleMismatch { - configured: self.config.role, - theirs, - }; - // Saying it twice is saying it once; saying two things is saying - // nothing that can be relied on. - if their_roles.any(|role| Some(role) != their_role) { - return self.fail(role_mismatch(None)); - } - let role_confirmed = match (self.config.role, their_role) { - (Some(configured), Some(theirs)) if theirs == configured.theirs() => true, - (Some(_), Some(theirs)) => return self.fail(role_mismatch(Some(theirs))), - (Some(_), None) if self.config.require_role => { - return self.fail(role_mismatch(None)); - } - // A peer that does not say is taken at our word for it; and - // what a peer says of itself means nothing where nothing is - // configured to hold it against. - (Some(_), None) | (None, _) => false, + let role_confirmed = match self.negotiate_role(&open) { + Ok(confirmed) => confirmed, + Err(e) => return self.fail(e), }; let ours = self.config.families.clone(); @@ -388,26 +363,7 @@ impl Fsm { if families.is_empty() { return self.fail(SessionError::NoCommonAddressFamily { ours, theirs }); } - - // ADD-PATH is in use in a direction, for a family the session - // carries, when one side offered to send and the other to receive - // (RFC 7911 §4). - let (mut add_path_receive, mut add_path_send) = - (FamilySet::default(), FamilySet::default()); - if let Some(ours) = self.config.add_path { - let theirs = open.capabilities().flat_map(|capability| match capability { - Capability::AddPath { entries } => entries.as_slice(), - _ => &[], - }); - for entry in theirs.filter(|entry| families.contains(entry.family())) { - if ours.receives() && entry.send_recv.sends() { - add_path_receive.insert(entry.family()); - } - if ours.sends() && entry.send_recv.receives() { - add_path_send.insert(entry.family()); - } - } - } + let (add_path_receive, add_path_send) = self.negotiate_add_path(&open, &families); // RFC 4724 §3: in use if both offered it. let graceful_restart = self @@ -522,6 +478,64 @@ impl Fsm { actions } + /// What the peer says it is to us, held against what is configured + /// (RFC 9234 §4.2): whether the two agree, or an error if they do not. + /// A peer that does not say is taken at our word for it, unless it is + /// required to; what a peer says of itself means nothing where nothing + /// is configured to hold it against. + fn negotiate_role(&self, open: &OpenMessage) -> Result { + let mut their_roles = open + .capabilities() + .filter_map(|capability| match capability { + Capability::BgpRole { role } => Some(*role), + _ => None, + }); + let their_role = their_roles.next(); + let mismatch = |theirs| SessionError::RoleMismatch { + configured: self.config.role, + theirs, + }; + // Saying it twice is saying it once; saying two things is saying + // nothing that can be relied on. + if their_roles.any(|role| Some(role) != their_role) { + return Err(mismatch(None)); + } + match (self.config.role, their_role) { + (Some(configured), Some(theirs)) if theirs == configured.theirs() => Ok(true), + (Some(_), Some(theirs)) => Err(mismatch(Some(theirs))), + (Some(_), None) if self.config.require_role => Err(mismatch(None)), + (Some(_), None) | (None, _) => Ok(false), + } + } + + /// The families the peer sends us several routes for a prefix in, and + /// those we send it several in: ADD-PATH is in use in a direction, for + /// a family the session carries, when one side offered to send and the + /// other to receive (RFC 7911 §4). + fn negotiate_add_path( + &self, + open: &OpenMessage, + families: &FamilySet, + ) -> (FamilySet, FamilySet) { + let (mut receive, mut send) = (FamilySet::default(), FamilySet::default()); + let Some(ours) = self.config.add_path else { + return (receive, send); + }; + let theirs = open.capabilities().flat_map(|capability| match capability { + Capability::AddPath { entries } => entries.as_slice(), + _ => &[], + }); + for entry in theirs.filter(|entry| families.contains(entry.family())) { + if ours.receives() && entry.send_recv.sends() { + receive.insert(entry.family()); + } + if ours.sends() && entry.send_recv.receives() { + send.insert(entry.family()); + } + } + (receive, send) + } + /// `notification` as it goes to the peer: where NOTIFICATIONs count as /// restarts (RFC 8538) and this one is `for_good`, as a Hard Reset, so /// that the peer drops our routes rather than keeping them. diff --git a/crates/server/src/holddown.rs b/crates/server/src/holddown.rs new file mode 100644 index 0000000..8fe1c05 --- /dev/null +++ b/crates/server/src/holddown.rs @@ -0,0 +1,174 @@ +//! Peers that are being kept down for a while: those that went over a +//! limit, ours or theirs, and those whose sessions keep failing. + +use std::collections::HashMap; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use crate::address::PeerAddress; +use crate::sync::IgnorePoison; + +/// Why a peer is being kept down. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HeldDown { + /// It went over its limit of routes. + OverLimit, + /// We were to announce it more routes than its export limit allows. + OverExportLimit, + /// Its sessions keep coming up and failing. + Flapping, +} + +impl HeldDown { + pub fn as_str(self) -> &'static str { + match self { + Self::OverLimit => "held down after going over its limit of routes", + Self::OverExportLimit => { + "held down after more routes were to be advertised to it than its export limit allows" + } + Self::Flapping => "held down because its sessions keep failing", + } + } +} + +/// A session that fails sooner than this after coming up counts as a +/// flap. +const FLAP_WITHIN: Duration = Duration::from_secs(60); +/// This many flaps in a row are let pass: a router being restarted, a +/// cable put back. +const FLAPS_ALLOWED: u32 = 2; +/// How long a peer is kept down after the first flap too many; twice as +/// long after each one more, up to `FLAP_HOLD_MAX`. +const FLAP_HOLD: Duration = Duration::from_secs(1); +const FLAP_HOLD_MAX: Duration = Duration::from_secs(300); + +/// Peers that are being kept down, and until when: those that went over +/// their limit, and those whose sessions keep failing. Shared by the +/// sessions, which put peers here, and by the dialer and the listeners, +/// which leave them alone meanwhile. +/// +/// Every session that comes up and goes costs a look at every route in +/// the table, and tells every other peer of the changes. A peer that +/// does so over and over, by fault or by design, is slowed down. +#[derive(Debug, Default)] +pub struct HoldDowns { + held: Mutex>, + /// How many times in a row each peer's session has failed soon after + /// coming up. + flaps: Mutex>, +} + +impl HoldDowns { + /// Keeps the peer at `address` down for `duration` from now. + pub fn hold(&self, address: PeerAddress, duration: Duration, why: HeldDown) { + let until = Instant::now() + duration; + let mut held = self.held.locked(); + // Not for less long than it already is. + if held + .get(&address) + .is_none_or(|(already, _)| *already < until) + { + held.insert(address, (until, why)); + } + } + + /// How much longer the peer at `address` is kept down, if it is, and + /// why. + pub fn remaining(&self, address: PeerAddress) -> Option<(Duration, HeldDown)> { + let mut held = self.held.locked(); + let (until, why) = *held.get(&address)?; + let remaining = until.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + held.remove(&address); + return None; + } + Some((remaining, why)) + } + + /// A session with the peer at `address` that had come up has ended, + /// `lasted` after it did; `failed` if not at our own wish. Returns for + /// how long the peer is kept down for it, if it is. + pub fn session_ended( + &self, + address: PeerAddress, + lasted: Duration, + failed: bool, + ) -> Option { + let mut flaps = self.flaps.locked(); + if !failed || lasted >= FLAP_WITHIN { + flaps.remove(&address); + return None; + } + let count = flaps.entry(address).or_insert(0); + *count += 1; + let over = count.checked_sub(FLAPS_ALLOWED + 1)?; + let hold = FLAP_HOLD + .checked_mul(1u32.checked_shl(over).unwrap_or(u32::MAX)) + .map_or(FLAP_HOLD_MAX, |hold| hold.min(FLAP_HOLD_MAX)); + drop(flaps); + self.hold(address, hold, HeldDown::Flapping); + Some(hold) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_peer_is_held_down_for_as_long_as_was_said() { + let held = HoldDowns::default(); + let address = PeerAddress::from("192.0.2.1".parse::().unwrap()); + let other = PeerAddress::from("192.0.2.2".parse::().unwrap()); + + // Two failures soon after coming up are let pass; from the third + // the peer is kept down, twice as long each time, up to a limit. + let flaps = HoldDowns::default(); + let soon = Duration::from_secs(5); + assert_eq!(flaps.session_ended(address, soon, true), None); + assert_eq!(flaps.session_ended(address, soon, true), None); + assert_eq!(flaps.session_ended(address, soon, true), Some(FLAP_HOLD)); + assert_eq!( + flaps.session_ended(address, soon, true), + Some(2 * FLAP_HOLD) + ); + assert_eq!( + flaps.remaining(address).map(|(_, why)| why), + Some(HeldDown::Flapping) + ); + assert_eq!(flaps.remaining(other), None, "each peer for itself"); + for _ in 0..40 { + flaps.session_ended(address, soon, true); + } + assert_eq!( + flaps.session_ended(address, soon, true), + Some(FLAP_HOLD_MAX) + ); + // A session that lasted, or that we ended ourselves, starts the + // count over. + assert_eq!(flaps.session_ended(address, FLAP_WITHIN, true), None); + assert_eq!(flaps.session_ended(address, soon, true), None); + assert_eq!(flaps.session_ended(address, soon, false), None); + assert_eq!(flaps.session_ended(address, soon, true), None); + assert_eq!(held.remaining(address), None); + + held.hold(address, Duration::from_secs(600), HeldDown::OverLimit); + let (remaining, why) = held.remaining(address).expect("held"); + assert!(remaining > Duration::from_secs(590), "{remaining:?}"); + assert_eq!(why, HeldDown::OverLimit); + assert_eq!(held.remaining(other), None); + // Not made shorter by a lesser reason. + held.hold(address, Duration::from_secs(1), HeldDown::Flapping); + let (remaining, why) = held.remaining(address).expect("held"); + assert!(remaining > Duration::from_secs(590), "{remaining:?}"); + assert_eq!(why, HeldDown::OverLimit); + + held.hold(other, Duration::ZERO, HeldDown::OverLimit); + assert_eq!(held.remaining(other), None); + assert_eq!( + held.held.locked().len(), + 1, + "a hold that is over is forgotten" + ); + } +} diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index bb58d69..addb0bb 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -8,6 +8,7 @@ pub mod error; pub mod export; pub mod family; mod fsm; +pub(crate) mod holddown; pub mod marks; pub mod metrics; mod panic; diff --git a/crates/server/src/policy.rs b/crates/server/src/policy.rs index 957842a..27f6bf6 100644 --- a/crates/server/src/policy.rs +++ b/crates/server/src/policy.rs @@ -26,8 +26,8 @@ use std::collections::{HashMap, HashSet}; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; -use std::sync::{Arc, LazyLock, Mutex}; -use std::time::{Duration, Instant}; +use std::sync::{Arc, LazyLock}; +use std::time::Duration; use bgp_packet::message::{ AS_TRANS, AsPathAttribute, LocalPrefPathAttribute, NlriNextHop, PathAttribute, @@ -36,9 +36,7 @@ use bgp_packet::open::BgpRole; use bgp_primitives::constants::IpVersion; use bgp_primitives::ip_prefix::IpPrefix; -use crate::address::PeerAddress; use crate::marks::{self, Marks, Patterns, Scrub}; -use crate::sync::IgnorePoison; /// The longest prefixes routed between networks: anything more specific is /// filtered nearly everywhere, and is usually a leak of internal routes. @@ -906,109 +904,6 @@ impl PrefixSet { } } -/// Why a peer is being kept down. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum HeldDown { - /// It went over its limit of routes. - OverLimit, - /// We were to announce it more routes than its export limit allows. - OverExportLimit, - /// Its sessions keep coming up and failing. - Flapping, -} - -impl HeldDown { - pub fn as_str(self) -> &'static str { - match self { - Self::OverLimit => "held down after going over its limit of routes", - Self::OverExportLimit => { - "held down after more routes were to be advertised to it than its export limit allows" - } - Self::Flapping => "held down because its sessions keep failing", - } - } -} - -/// A session that fails sooner than this after coming up counts as a -/// flap. -const FLAP_WITHIN: Duration = Duration::from_secs(60); -/// This many flaps in a row are let pass: a router being restarted, a -/// cable put back. -const FLAPS_ALLOWED: u32 = 2; -/// How long a peer is kept down after the first flap too many; twice as -/// long after each one more, up to `FLAP_HOLD_MAX`. -const FLAP_HOLD: Duration = Duration::from_secs(1); -const FLAP_HOLD_MAX: Duration = Duration::from_secs(300); - -/// Peers that are being kept down, and until when: those that went over -/// their limit, and those whose sessions keep failing. Shared by the -/// sessions, which put peers here, and by the dialer and the listeners, -/// which leave them alone meanwhile. -/// -/// Every session that comes up and goes costs a look at every route in -/// the table, and tells every other peer of the changes. A peer that -/// does so over and over, by fault or by design, is slowed down. -#[derive(Debug, Default)] -pub struct HoldDowns { - held: Mutex>, - /// How many times in a row each peer's session has failed soon after - /// coming up. - flaps: Mutex>, -} - -impl HoldDowns { - /// Keeps the peer at `address` down for `duration` from now. - pub fn hold(&self, address: PeerAddress, duration: Duration, why: HeldDown) { - let until = Instant::now() + duration; - let mut held = self.held.locked(); - // Not for less long than it already is. - if held - .get(&address) - .is_none_or(|(already, _)| *already < until) - { - held.insert(address, (until, why)); - } - } - - /// How much longer the peer at `address` is kept down, if it is, and - /// why. - pub fn remaining(&self, address: PeerAddress) -> Option<(Duration, HeldDown)> { - let mut held = self.held.locked(); - let (until, why) = *held.get(&address)?; - let remaining = until.saturating_duration_since(Instant::now()); - if remaining.is_zero() { - held.remove(&address); - return None; - } - Some((remaining, why)) - } - - /// A session with the peer at `address` that had come up has ended, - /// `lasted` after it did; `failed` if not at our own wish. Returns for - /// how long the peer is kept down for it, if it is. - pub fn session_ended( - &self, - address: PeerAddress, - lasted: Duration, - failed: bool, - ) -> Option { - let mut flaps = self.flaps.locked(); - if !failed || lasted >= FLAP_WITHIN { - flaps.remove(&address); - return None; - } - let count = flaps.entry(address).or_insert(0); - *count += 1; - let over = count.checked_sub(FLAPS_ALLOWED + 1)?; - let hold = FLAP_HOLD - .checked_mul(1u32.checked_shl(over).unwrap_or(u32::MAX)) - .map_or(FLAP_HOLD_MAX, |hold| hold.min(FLAP_HOLD_MAX)); - drop(flaps); - self.hold(address, hold, HeldDown::Flapping); - Some(hold) - } -} - #[cfg(test)] mod tests { use bgp_packet::message::{AsPathSegment, CommunitiesPathAttribute}; @@ -1744,63 +1639,6 @@ mod tests { ); } - #[test] - fn a_peer_is_held_down_for_as_long_as_was_said() { - let held = HoldDowns::default(); - let address = PeerAddress::from("192.0.2.1".parse::().unwrap()); - let other = PeerAddress::from("192.0.2.2".parse::().unwrap()); - - // Two failures soon after coming up are let pass; from the third - // the peer is kept down, twice as long each time, up to a limit. - let flaps = HoldDowns::default(); - let soon = Duration::from_secs(5); - assert_eq!(flaps.session_ended(address, soon, true), None); - assert_eq!(flaps.session_ended(address, soon, true), None); - assert_eq!(flaps.session_ended(address, soon, true), Some(FLAP_HOLD)); - assert_eq!( - flaps.session_ended(address, soon, true), - Some(2 * FLAP_HOLD) - ); - assert_eq!( - flaps.remaining(address).map(|(_, why)| why), - Some(HeldDown::Flapping) - ); - assert_eq!(flaps.remaining(other), None, "each peer for itself"); - for _ in 0..40 { - flaps.session_ended(address, soon, true); - } - assert_eq!( - flaps.session_ended(address, soon, true), - Some(FLAP_HOLD_MAX) - ); - // A session that lasted, or that we ended ourselves, starts the - // count over. - assert_eq!(flaps.session_ended(address, FLAP_WITHIN, true), None); - assert_eq!(flaps.session_ended(address, soon, true), None); - assert_eq!(flaps.session_ended(address, soon, false), None); - assert_eq!(flaps.session_ended(address, soon, true), None); - assert_eq!(held.remaining(address), None); - - held.hold(address, Duration::from_secs(600), HeldDown::OverLimit); - let (remaining, why) = held.remaining(address).expect("held"); - assert!(remaining > Duration::from_secs(590), "{remaining:?}"); - assert_eq!(why, HeldDown::OverLimit); - assert_eq!(held.remaining(other), None); - // Not made shorter by a lesser reason. - held.hold(address, Duration::from_secs(1), HeldDown::Flapping); - let (remaining, why) = held.remaining(address).expect("held"); - assert!(remaining > Duration::from_secs(590), "{remaining:?}"); - assert_eq!(why, HeldDown::OverLimit); - - held.hold(other, Duration::ZERO, HeldDown::OverLimit); - assert_eq!(held.remaining(other), None); - assert_eq!( - held.held.locked().len(), - 1, - "a hold that is over is forgotten" - ); - } - /// The quick look at a prefix's first octets never lets through what /// the list has. fn assert_quick_check_agrees(prefix: IpPrefix) { diff --git a/crates/server/src/server.rs b/crates/server/src/server.rs index 7c57519..d487d26 100644 --- a/crates/server/src/server.rs +++ b/crates/server/src/server.rs @@ -25,9 +25,10 @@ use crate::config::{Config, PeerConfig}; use crate::connection::{PeerSettings, ServerContext, off_thread, run_session}; use crate::dialer::spawn_dialers; use crate::error::ServerError; +use crate::holddown::HoldDowns; use crate::metrics::{CONNECTIONS_REFUSED, PEER_MAX_PREFIXES}; use crate::panic; -use crate::policy::{ExportLimits, HoldDowns}; +use crate::policy::ExportLimits; use crate::reach; use crate::refusals::RefusalLog; use crate::restart::{Leave, Restarts}; @@ -178,32 +179,9 @@ impl BgpServer { /// Runs the server. Returns only if something it cannot do without has /// failed, after closing its sessions. - pub async fn run(self) -> Result<(), ServerError> { + pub async fn run(mut self) -> Result<(), ServerError> { crate::metrics::init(); - info!( - router_id = %self.config.router_id, - local_asn = self.config.local_asn, - peers = self.config.peers.len(), - "BGP server started" - ); - - for peer in &self.config.peers { - if let Some(limit) = peer.import.max_prefixes { - PEER_MAX_PREFIXES - .with_label_values(&[&peer.name]) - .set(limit.into()); - } else { - warn!( - peer = %peer.name, - "No limit on the routes taken from this peer (import.max_prefixes)" - ); - } - } - warn_about_export_limits(&self.config); - if self.config.accept_unconfigured_peers { - warn!("Taking sessions and routes from any host (accept_unconfigured_peers)"); - } - + log_startup(&self.config); let context = Arc::new(ServerContext { router_id: self.config.router_id, local_asn: self.config.local_asn, @@ -226,43 +204,7 @@ impl BgpServer { } }); } - - // The tasks the server cannot do without. None of them ends while - // 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( - self.local_rib.clone(), - Arc::clone(&self.rib), - self.registry.clone(), - self.config.local_asn, - self.config.control, - self.config.cluster_id, - self.restarts.convergence().watch(), - ), - )); - tasks.push(named("sweeper", spawn_sweeper(Arc::clone(&self.rib)))); - if let (Some(rpki), Some(client)) = (self.rpki.clone(), self.rpki_client) { - tasks.push(named("RPKI client", client)); - tasks.push(named( - "revalidator", - spawn_revalidator(rpki, Arc::clone(&self.rib), self.registry.clone()), - )); - } - if let Some(interfaces) = context.interfaces.clone() { - tasks.push(named( - "next hop revalidator", - reach::spawn_revalidator(interfaces, Arc::clone(&self.rib), self.registry.clone()), - )); - } - for peer in self.config.peers.iter().filter(|peer| peer.passive) { - info!(peer = %peer.name, endpoint = %peer.endpoint, "Waiting for the peer to connect (passive)"); - } - for dialer in spawn_dialers(&self.config.peers, &context, self.config.connect_retry) { - tasks.push(named("dialer", dialer)); - } + let mut tasks = self.spawn_tasks(&context); let mut kernel_task = self.kernel_task; let connections = ConnectionCounts::default(); @@ -308,6 +250,46 @@ impl BgpServer { accepting.push(accept_one(listen_addr, listener)); } } + + /// Starts the tasks the server cannot do without. None of them ends + /// while the server runs. The advertiser comes first: it has to be + /// listening before a dialer brings up the first session. + fn spawn_tasks(&mut self, context: &Arc) -> FuturesUnordered { + let tasks = FuturesUnordered::new(); + tasks.push(named( + "advertiser", + spawn_advertiser( + self.local_rib.clone(), + Arc::clone(&self.rib), + self.registry.clone(), + self.config.local_asn, + self.config.control, + self.config.cluster_id, + self.restarts.convergence().watch(), + ), + )); + tasks.push(named("sweeper", spawn_sweeper(Arc::clone(&self.rib)))); + if let (Some(rpki), Some(client)) = (self.rpki.clone(), self.rpki_client.take()) { + tasks.push(named("RPKI client", client)); + tasks.push(named( + "revalidator", + spawn_revalidator(rpki, Arc::clone(&self.rib), self.registry.clone()), + )); + } + if let Some(interfaces) = context.interfaces.clone() { + tasks.push(named( + "next hop revalidator", + reach::spawn_revalidator(interfaces, Arc::clone(&self.rib), self.registry.clone()), + )); + } + for peer in self.config.peers.iter().filter(|peer| peer.passive) { + info!(peer = %peer.name, endpoint = %peer.endpoint, "Waiting for the peer to connect (passive)"); + } + for dialer in spawn_dialers(&self.config.peers, context, self.config.connect_retry) { + tasks.push(named("dialer", dialer)); + } + tasks + } } /// Resolves when the task that keeps the kernel's routes has ended, which @@ -380,6 +362,32 @@ async fn leave_sessions(registry: &SessionRegistry, leave: Leave) { } } +/// Says what the configuration is, and what it leaves undone. +fn log_startup(config: &Config) { + info!( + router_id = %config.router_id, + local_asn = config.local_asn, + peers = config.peers.len(), + "BGP server started" + ); + for peer in &config.peers { + if let Some(limit) = peer.import.max_prefixes { + PEER_MAX_PREFIXES + .with_label_values(&[&peer.name]) + .set(limit.into()); + } else { + warn!( + peer = %peer.name, + "No limit on the routes taken from this peer (import.max_prefixes)" + ); + } + } + warn_about_export_limits(config); + if config.accept_unconfigured_peers { + warn!("Taking sessions and routes from any host (accept_unconfigured_peers)"); + } +} + /// Says what the export limits leave undone: the peers they are for that /// have none, and the routes of our own that they leave out, which is /// likely a mistake in one or the other.