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

Server: the long functions in named steps

Browse Source 
Five functions did several things each in one body of 120 to 190 lines.
Each is now a list of steps with a name and a doc comment of their own,
with nothing changed in what they do:

- run_session: hold_down_after, keep_for and release_routes.
- Session::establish: session_handle, validation and route_source. A
  session keeps the peer's settings as one PeerSettings rather than
  eight fields copied out of it one by one; the sweeps' two channel ends
  are named fields, not .0 and .1; whether the peer can be asked to send
  its routes again is a bool, not a copy of the families.
- Fsm::on_open: negotiate_role and negotiate_add_path.
- BgpServer::run: log_startup and spawn_tasks.
- Config::from_proto: parse_listen_endpoints, parse_peers, check_roles,
  parse_originate and Ours::tell_peers, with the three per-peer defaults
  in one PeerDefaults.
- Egress::outgoing: may_go, rewrite and next_hop_for.
- Exporter::follow: Learned::drain, judge and flush, with the three
  things called "announced" told apart (own_announced, learned.announced,
  batches) and the verdicts given a type.

The hold-downs are a module of their own, holddown.rs; policy.rs is
about policy again.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Rayhaan JaufeerallyandClaude Fable 5.1 committed 2026-10-06 11:32:20 +02:00
1 parent 724bb99f11
commit 0829f873f2
9 files changed
+865 -672
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/server/src/advertise.rs
+86 -56
View File
@@ -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<Notify>,
/// 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<IpPrefix>,
/// 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<IpPrefix>,
/// The families to send End-of-RIB markers for, until they are sent;
/// `None` if graceful restart was not agreed.
end_of_rib: Option<FamilySet>,
@@ -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<IpPrefix, Option<Arc<Dressed>>>;
/// 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<HashMap<IpPrefix, Option<Arc<Dressed>>>>,
gathered: Option<Verdicts>,
permit: Option<OwnedSemaphorePermit>,
/// The prefixes announced to the peer and not withdrawn since.
announced: HashSet<IpPrefix>,
@@ -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<News> {
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<News>) -> 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<Nlri>; 2] = Default::default();
let mut announced: HashMap<(usize, IpVersion), (Arc<Dressed>, Vec<Nlri>)> = HashMap::new();
let mut batches: HashMap<(usize, IpVersion), (Arc<Dressed>, Vec<Nlri>)> = 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<IpPrefix> = routes.iter().map(|route| route.prefix).collect();
let gone: Vec<IpPrefix> = self.announced.difference(&current).copied().collect();
let gone: Vec<IpPrefix> = self.own_announced.difference(&current).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(());
}
crates/server/src/config.rs
+190 -134
View File
@@ -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::<Result<_, _>>()?
};
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<PeerConfig> = 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<LocalRoute> = 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<Vec<SocketAddr>, 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<ExportPolicy>,
export_limits: Option<Arc<ExportLimits>>,
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<Peer>,
local_asn: u32,
defaults: &PeerDefaults,
) -> Result<Vec<PeerConfig>, ConfigError> {
let mut parsed: Vec<PeerConfig> = 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<Vec<LocalRoute>, ConfigError> {
let mut originate: Vec<LocalRoute> = 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<Ipv4Addr> {
@@ -360,12 +421,7 @@ fn parse_export_limits(
Ok((limits != ExportLimits::default()).then_some(limits))
}
fn parse_peer(
peer: Peer,
export: &Arc<ExportPolicy>,
export_limits: Option<&Arc<ExportLimits>>,
graceful_restart: GracefulRestartConfig,
) -> Result<PeerConfig, PeerProblem> {
fn parse_peer(peer: Peer, defaults: &PeerDefaults) -> Result<PeerConfig, PeerProblem> {
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,
crates/server/src/connection.rs
+210 -175
View File
@@ -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<IpVersion>, 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<Stale>,
/// Where the sweeps of stale routes say what they did.
swept: (
mpsc::UnboundedSender<Outcome>,
mpsc::UnboundedReceiver<Outcome>,
),
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<Outcome>,
reported: mpsc::UnboundedReceiver<Outcome>,
}
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<ServerContext>,
/// Configured name of the peer, if it is a configured one.
peer_name: Option<String>,
/// Which of the peer's routes are taken.
import: Arc<ImportPolicy>,
/// What the peer is to us, if that is known.
role: Option<PeerRole>,
/// What the peer is told, and how.
export: Arc<ExportPolicy>,
/// Limits on what the peer is told, if it is one they are for.
export_limits: Option<Arc<ExportLimits>>,
/// 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<PeerRefusals>,
/// Graceful restart with the peer, as configured.
graceful_restart: GracefulRestartConfig,
/// Says when the daemon is leaving, and how.
leave: watch::Receiver<Option<Leave>>,
}
@@ -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<Message>,
) -> 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<Validation> {
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<Established>) -> 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,
crates/server/src/export.rs
+69 -32
View File
@@ -161,26 +161,46 @@ impl Egress<'_> {
source: Source,
routes: IpVersion,
) -> Option<Outgoing> {
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<PathAttribute>,
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<NlriNextHop>,
) -> Option<NlriNextHop> {
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),
)
}
}
crates/server/src/fsm.rs
+62 -48
View File
@@ -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<bool, SessionError> {
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.
crates/server/src/holddown.rs
+174
View File
@@ -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<HashMap<PeerAddress, (Instant, HeldDown)>>,
/// How many times in a row each peer's session has failed soon after
/// coming up.
flaps: Mutex<HashMap<PeerAddress, u32>>,
}
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<Duration> {
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::<std::net::IpAddr>().unwrap());
let other = PeerAddress::from("192.0.2.2".parse::<std::net::IpAddr>().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"
);
}
}
crates/server/src/lib.rs
+1
View File
@@ -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;
crates/server/src/policy.rs
+2 -164
View File
@@ -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<HashMap<PeerAddress, (Instant, HeldDown)>>,
/// How many times in a row each peer's session has failed soon after
/// coming up.
flaps: Mutex<HashMap<PeerAddress, u32>>,
}
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<Duration> {
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::<std::net::IpAddr>().unwrap());
let other = PeerAddress::from("192.0.2.2".parse::<std::net::IpAddr>().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) {
crates/server/src/server.rs
+71 -63
View File
@@ -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<NamedTask> = 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<ServerContext>) -> FuturesUnordered<NamedTask> {
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.