diff --git a/bfd/Cargo.toml b/bfd/Cargo.toml index ccb0bb70..bffdcfc9 100644 --- a/bfd/Cargo.toml +++ b/bfd/Cargo.toml @@ -3,6 +3,9 @@ name = "bfd" version = "0.1.0" edition = "2024" +[features] +test-support = [] + [dependencies] anyhow.workspace = true mg-common.workspace = true diff --git a/bfd/src/daemon.rs b/bfd/src/daemon.rs index e85b5eb0..8d45acf6 100644 --- a/bfd/src/daemon.rs +++ b/bfd/src/daemon.rs @@ -29,6 +29,11 @@ impl Daemon { Self::with_dispatcher(Dispatcher::new(), log) } + #[cfg(feature = "test-support")] + pub fn new_for_test(log: Logger) -> Self { + Self::with_dispatcher(Dispatcher::new_for_test(), log) + } + // Non-public method to allow construction with a custom dispatcher. // // This is used by tests when they want to use a `Dispatcher` with a custom diff --git a/bfd/src/dispatcher.rs b/bfd/src/dispatcher.rs index 16e59701..fa1cd2e2 100644 --- a/bfd/src/dispatcher.rs +++ b/bfd/src/dispatcher.rs @@ -66,6 +66,11 @@ impl Dispatcher { Self::with_backend(Arc::new(TokioUdpBinder)) } + #[cfg(feature = "test-support")] + pub(crate) fn new_for_test() -> Self { + Self::with_backend(Arc::new(NoopListenerBackend)) + } + fn with_backend(backend: Arc) -> Self { Self { peer_to_listen_addr: HashMap::default(), @@ -276,6 +281,21 @@ trait ListenerBackend: Send + Sync + 'static { ) -> Result, AddPeerError>; } +#[cfg(feature = "test-support")] +struct NoopListenerBackend; + +#[cfg(feature = "test-support")] +impl ListenerBackend for NoopListenerBackend { + fn spawn( + &self, + _listen_addr: SocketAddr, + _sessions: SharedSessions, + _log: Logger, + ) -> Result, AddPeerError> { + Ok(tokio::spawn(std::future::pending())) + } +} + /// Production [`ListenerBackend`]: binds a real UDP socket and spawns a /// [`ListenerTask`] to read from it. struct TokioUdpBinder; diff --git a/mg-api-types/versions/src/bfd_nonzero_detect_mult/bfd.rs b/mg-api-types/versions/src/bfd_nonzero_detect_mult/bfd.rs index 273966b1..b6d6271d 100644 --- a/mg-api-types/versions/src/bfd_nonzero_detect_mult/bfd.rs +++ b/mg-api-types/versions/src/bfd_nonzero_detect_mult/bfd.rs @@ -9,7 +9,9 @@ use serde::{Deserialize, Serialize}; use std::net::IpAddr; use std::num::NonZeroU8; -#[derive(Debug, Copy, Clone, Deserialize, Serialize, JsonSchema)] +#[derive( + Debug, Copy, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema, +)] pub struct BfdPeerConfig { /// Address of the peer to add. pub peer: IpAddr, diff --git a/mg-api-types/versions/src/initial/bfd.rs b/mg-api-types/versions/src/initial/bfd.rs index a76b48ca..d990cc96 100644 --- a/mg-api-types/versions/src/initial/bfd.rs +++ b/mg-api-types/versions/src/initial/bfd.rs @@ -7,7 +7,9 @@ use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use std::net::IpAddr; -#[derive(Debug, Copy, Clone, Deserialize, Serialize, JsonSchema)] +#[derive( + Debug, Copy, Clone, PartialEq, Eq, Deserialize, Serialize, JsonSchema, +)] pub struct BfdPeerConfig { /// Address of the peer to add. pub peer: IpAddr, @@ -21,7 +23,9 @@ pub struct BfdPeerConfig { pub mode: SessionMode, } -#[derive(Debug, Copy, Clone, Serialize, Deserialize, JsonSchema)] +#[derive( + Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema, +)] pub enum SessionMode { SingleHop, MultiHop, diff --git a/mgd/Cargo.toml b/mgd/Cargo.toml index 790ce55e..ef62369d 100644 --- a/mgd/Cargo.toml +++ b/mgd/Cargo.toml @@ -41,6 +41,7 @@ socket2.workspace = true [dev-dependencies] tempfile = "3" proptest.workspace = true +bfd = { workspace = true, features = ["test-support"] } [features] default = ["mg-lower"] diff --git a/mgd/src/admin.rs b/mgd/src/admin.rs index 3f8be363..d73bc1a0 100644 --- a/mgd/src/admin.rs +++ b/mgd/src/admin.rs @@ -123,21 +123,26 @@ impl MgAdminApi for MgAdminApiImpl { async fn get_bfd_peers( ctx: RequestContext, ) -> Result>, HttpError> { - bfd_admin::get_bfd_peers(ctx).await + ctx.context().bfd.get_peers().map(HttpResponseOk) } async fn add_bfd_peer( ctx: RequestContext, request: TypedBody, ) -> Result { - bfd_admin::add_bfd_peer(ctx, request).await + let ctx = ctx.context(); + ctx.bfd.add_new_peer(ctx.db.clone(), request.into_inner())?; + Ok(HttpResponseUpdatedNoContent()) } async fn remove_bfd_peer( ctx: RequestContext, params: Path, ) -> Result { - bfd_admin::remove_bfd_peer(ctx, params).await + let ctx = ctx.context(); + let peer = params.into_inner().addr; + ctx.bfd.remove_peer(ctx.db.clone(), peer).await?; + Ok(HttpResponseUpdatedNoContent()) } async fn read_routers( diff --git a/mgd/src/bfd_admin.rs b/mgd/src/bfd_admin.rs index 5cebc246..2e9c9da0 100644 --- a/mgd/src/bfd_admin.rs +++ b/mgd/src/bfd_admin.rs @@ -2,27 +2,24 @@ // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at https://mozilla.org/MPL/2.0/. -use crate::admin::HandlerContext; use anyhow::Result; -use bfd::AddPeerError; use bfd::Daemon; -use dropshot::{ - ClientErrorStatusCode, HttpError, HttpResponseOk, - HttpResponseUpdatedNoContent, Path, RequestContext, TypedBody, -}; +use bfd::Session; +use bfd::SessionCounters; +use dropshot::HttpError; use mg_api_types::bfd::BfdPeerConfig; use mg_api_types::bfd::BfdPeerInfo; -use mg_api_types::bfd::DeleteBfdPeerPathParams; use mg_common::lock; use slog::Logger; use slog_error_chain::InlineErrorChain; +use std::net::{IpAddr, SocketAddr}; use std::sync::{Arc, Mutex}; /// Context for Dropshot requests. #[derive(Clone)] pub struct BfdContext { /// The underlying deamon being run. - pub(crate) daemon: Arc>, + daemon: Arc>, } impl BfdContext { @@ -31,95 +28,274 @@ impl BfdContext { daemon: Arc::new(Mutex::new(Daemon::new(log.clone()))), } } -} -/// Get all the peers and their associated BFD state. Peers are identified by IP -/// address. -pub(crate) async fn get_bfd_peers( - ctx: RequestContext>, -) -> Result>, HttpError> { - let mut result = Vec::new(); - let daemon = lock!(ctx.context().bfd.daemon); - for (addr, session) in daemon.sessions_iter() { - result.push(BfdPeerInfo { + #[cfg(test)] + fn new_for_test(log: Logger) -> Self { + Self { + daemon: Arc::new(Mutex::new(Daemon::new_for_test(log))), + } + } + + /// Helper function for constructing a BfdPeerInfo + fn peer_info( + peer: &IpAddr, + session: &Session, + listener: Option, + ) -> Result { + let listen = listener + .ok_or(HttpError::for_internal_error(format!( + "no listener for {peer}" + )))? + .ip(); + Ok(BfdPeerInfo { config: BfdPeerConfig { - peer: *addr, + peer: *peer, + listen, required_rx: session.required_rx_micros(), detection_threshold: session.detection_threshold(), - listen: daemon - .listen_addr_for_peer(addr) - .ok_or(HttpError::for_internal_error(format!( - "no listener for {addr}" - )))? - .ip(), mode: session.mode(), }, state: session.state(), - }); + }) } - Ok(HttpResponseOk(result)) -} + /// Get all BFD peers and their associated state. + /// Peers are identified by IP address. + pub fn get_peers(&self) -> Result, HttpError> { + let mut result = Vec::new(); + let daemon = lock!(self.daemon); + for (addr, session) in daemon.sessions_iter() { + let info = Self::peer_info( + addr, + session, + daemon.listen_addr_for_peer(addr), + )?; + result.push(info); + } -/// Add a new peer to the daemon. A session for the specified peer will start -/// immediately. -pub(crate) async fn add_bfd_peer( - ctx: RequestContext>, - request: TypedBody, -) -> Result { - add_peer(ctx.context().clone(), request.into_inner())?; - Ok(HttpResponseUpdatedNoContent()) -} + Ok(result) + } -pub(crate) fn add_peer( - ctx: Arc, - rq: BfdPeerConfig, -) -> Result<(), HttpError> { - let mut daemon = lock!(ctx.bfd.daemon); - daemon - .add_peer(ctx.db.clone(), rq.into()) - .map_err(|err| match err { - AddPeerError::PeerExists(_) => HttpError::for_client_error( - None, - ClientErrorStatusCode::CONFLICT, - InlineErrorChain::new(&err).to_string(), - ), - AddPeerError::Bind { .. } - | AddPeerError::SetSocketNonBlocking(_) - | AddPeerError::StdToTokio(_) => HttpError::for_internal_error( - InlineErrorChain::new(&err).to_string(), - ), - }) + /// Restore persisted BFD peers to the running daemon. + /// + /// This reads the peer configs from the DB itself so callers cannot start + /// an unpersisted peer through the production API. + pub fn restore_db_peers( + &self, + db: rdb::Db, + ) -> Result<(), crate::error::Error> { + let mut daemon = lock!(self.daemon); + for config in db.get_bfd_neighbors()? { + daemon.add_peer(db.clone(), config.into())?; + } + Ok(()) + } + + /// Add a new BFD peer to persistent config and the running daemon. + /// + /// This first tries to update the persistent DB with the BFD peer config, + /// starting its runtime only upon success; if the DB update fails, an error + /// is returned. A rollback of the DB update is attempted if the subsequent + /// runtime update fails, returning an internal error upon rollback failure. + pub fn add_new_peer( + &self, + db: rdb::Db, + rq: BfdPeerConfig, + ) -> Result<(), HttpError> { + let mut daemon = lock!(self.daemon); + + db.clone() + .add_bfd_neighbor(rq) + .map_err(crate::error::Error::Db)?; + + if let Err(e) = daemon.add_peer(db.clone(), rq.into()) { + db.remove_bfd_neighbor(rq.peer).map_err(|e| { + HttpError::for_internal_error( + InlineErrorChain::new(&e).to_string(), + ) + })?; + return Err(crate::error::Error::Bfd(e).into()); + } + + Ok(()) + } + + /// Remove a BFD peer from the persistent DB and the running daemon. + /// The associated peer session will be stopped immediately. + pub async fn remove_peer( + &self, + db: rdb::Db, + peer: IpAddr, + ) -> Result<(), HttpError> { + let handle = { + let mut daemon = lock!(self.daemon); + db.remove_bfd_neighbor(peer).map_err(|e| { + HttpError::for_internal_error( + InlineErrorChain::new(&e).to_string(), + ) + })?; + daemon.remove_peer(peer) + }; + + if let Some(handle) = handle { + // If this was the last peer associated with a given local listening + // address, wait for the listening socket to be closed (allowing a + // caller to add a new peer at the same listening address once this + // returns). + // + // We've already unlocked the `bfd.daemon`, so it's possible a + // _concurrent_ request for the same listen address we're shutting down + // here could fail, but that's inherently racy: we can only guarantee + // that a client waiting for this remove to complete is able to add a + // new peer at the same listening address. + handle.shutdown().await; + } + + Ok(()) + } + + pub fn session_counters(&self) -> Vec<(IpAddr, Arc)> { + lock!(self.daemon) + .sessions_iter() + .map(|(ip, session)| (*ip, Arc::clone(session.counters()))) + .collect() + } } -/// Remove the specified peer from the daemon. The associated peer session will -/// be stopped immediately. -pub(crate) async fn remove_bfd_peer( - ctx: RequestContext>, - params: Path, -) -> Result { - let rq = params.into_inner(); - let listener_shutdown_handle = ctx - .context() - .bfd - .daemon - .lock() - .unwrap() - .remove_peer(rq.addr); - - if let Some(handle) = listener_shutdown_handle { - // If this was the last peer associated with a given local listening - // address, wait for the listening socket to be closed (allowing a - // caller to add a new peer at the same listening address once this - // returns). - // - // We've already unlocked the `bfd.daemon`, so it's possible a - // _concurrent_ request for the same listen address we're shutting down - // here could fail, but that's inherently racy: we can only guarantee - // that a client waiting for this remove to complete is able to add a - // new peer at the same listening address. - handle.shutdown().await; - } - - Ok(HttpResponseUpdatedNoContent {}) +#[cfg(test)] +mod tests { + use super::BfdContext; + use mg_api_types::bfd::{BfdPeerConfig, SessionMode}; + use mg_common::lock; + use slog::Logger; + use std::net::Ipv4Addr; + use std::num::NonZeroU8; + use tempfile::TempDir; + + fn setup() -> (BfdContext, rdb::Db, TempDir) { + let temp_dir = TempDir::new().unwrap(); + let log = Logger::root(slog::Discard, slog::o!()); + let db = rdb::Db::new(temp_dir.path().to_str().unwrap(), log.clone()) + .unwrap(); + (BfdContext::new_for_test(log), db, temp_dir) + } + + fn peer_config(host: u8, mode: SessionMode) -> BfdPeerConfig { + BfdPeerConfig { + peer: Ipv4Addr::new(127, 0, 1, host).into(), + listen: Ipv4Addr::LOCALHOST.into(), + required_rx: 100_000, + detection_threshold: NonZeroU8::new(3).unwrap(), + mode, + } + } + + fn start_runtime_peer( + bfd: &BfdContext, + db: rdb::Db, + config: BfdPeerConfig, + ) { + lock!(bfd.daemon).add_peer(db, config.into()).unwrap(); + } + + fn persisted_and_running_configs( + bfd: &BfdContext, + db: &rdb::Db, + ) -> (Vec, Vec) { + let persisted = db.get_bfd_neighbors().unwrap(); + let running = bfd + .get_peers() + .unwrap() + .into_iter() + .map(|peer| peer.config) + .collect(); + (persisted, running) + } + + #[tokio::test] + async fn add_peer_writes_db() { + let (bfd, db, _temp_dir) = setup(); + let config = peer_config(1, SessionMode::SingleHop); + + bfd.add_new_peer(db.clone(), config).unwrap(); + + assert_eq!(db.get_bfd_neighbors().unwrap(), vec![config]); + bfd.remove_peer(db, config.peer).await.unwrap(); + } + + #[tokio::test] + async fn remove_peer_deletes_db_entry() { + let (bfd, db, _temp_dir) = setup(); + let config = peer_config(2, SessionMode::MultiHop); + db.add_bfd_neighbor(config).unwrap(); + start_runtime_peer(&bfd, db.clone(), config); + + bfd.remove_peer(db.clone(), config.peer).await.unwrap(); + + assert!(db.get_bfd_neighbors().unwrap().is_empty()); + } + + #[tokio::test] + async fn add_peer_rolls_back_db_on_runtime_error() { + let (bfd, db, _temp_dir) = setup(); + let config = peer_config(3, SessionMode::SingleHop); + start_runtime_peer(&bfd, db.clone(), config); + + bfd.add_new_peer(db.clone(), config).unwrap_err(); + + assert_eq!( + persisted_and_running_configs(&bfd, &db), + (vec![], vec![config]), + ); + bfd.remove_peer(db, config.peer).await.unwrap(); + } + + #[tokio::test] + async fn add_peer_skips_runtime_on_db_conflict() { + let (bfd, db, _temp_dir) = setup(); + let config = peer_config(4, SessionMode::SingleHop); + db.add_bfd_neighbor(config).unwrap(); + + bfd.add_new_peer(db.clone(), config).unwrap_err(); + + assert_eq!( + persisted_and_running_configs(&bfd, &db), + (vec![config], vec![]), + ); + bfd.remove_peer(db, config.peer).await.unwrap(); + } + + #[tokio::test] + async fn restore_peers_leaves_db_unchanged() { + let (bfd, db, _temp_dir) = setup(); + let config = peer_config(5, SessionMode::SingleHop); + db.add_bfd_neighbor(config).unwrap(); + + bfd.restore_db_peers(db.clone()).unwrap(); + + assert_eq!(db.get_bfd_neighbors().unwrap(), vec![config]); + bfd.remove_peer(db, config.peer).await.unwrap(); + } + + #[tokio::test] + async fn remove_peer_cleans_up_db_only_peer() { + let (bfd, db, _temp_dir) = setup(); + let config = peer_config(6, SessionMode::SingleHop); + db.add_bfd_neighbor(config).unwrap(); + + bfd.remove_peer(db.clone(), config.peer).await.unwrap(); + + assert_eq!(persisted_and_running_configs(&bfd, &db), (vec![], vec![])); + } + + #[tokio::test] + async fn remove_peer_cleans_up_runtime_only_peer() { + let (bfd, db, _temp_dir) = setup(); + let config = peer_config(7, SessionMode::SingleHop); + start_runtime_peer(&bfd, db.clone(), config); + + bfd.remove_peer(db.clone(), config.peer).await.unwrap(); + + assert_eq!(persisted_and_running_configs(&bfd, &db), (vec![], vec![])); + } } diff --git a/mgd/src/error.rs b/mgd/src/error.rs index 23677daa..6992af64 100644 --- a/mgd/src/error.rs +++ b/mgd/src/error.rs @@ -16,6 +16,9 @@ pub enum Error { #[error("not found: {0}")] NotFound(String), + #[error("bfd error: {0}")] + Bfd(#[from] bfd::AddPeerError), + #[error("bgp error: {0}")] Bgp(#[from] bgp::error::Error), @@ -47,6 +50,20 @@ impl From for HttpError { ClientErrorStatusCode::CONFLICT, ), Error::NotFound(_) => Self::for_not_found(None, value.to_string()), + Error::Bfd(ref err) => match err { + bfd::AddPeerError::PeerExists(ip_addr) => { + Self::for_client_error( + Some(err.to_string()), + ClientErrorStatusCode::CONFLICT, + format!("bfd peer {ip_addr} already exists"), + ) + } + bfd::AddPeerError::Bind { .. } + | bfd::AddPeerError::SetSocketNonBlocking(_) + | bfd::AddPeerError::StdToTokio(_) => { + Self::for_internal_error(value.to_string()) + } + }, Error::Bgp(ref err) => match err { bgp::error::Error::PeerExists(_) => { Self::for_client_error_with_status( diff --git a/mgd/src/main.rs b/mgd/src/main.rs index fd03886e..77036c44 100644 --- a/mgd/src/main.rs +++ b/mgd/src/main.rs @@ -9,7 +9,6 @@ use crate::log::dlog; use bgp::connection_tcp::{BgpConnectionTcp, BgpListenerTcp}; use camino::Utf8PathBuf; use clap::{Parser, Subcommand}; -use mg_api_types::bfd::BfdPeerConfig; use mg_api_types::bgp::config::{ BgpPeerParameters, Ipv4UnicastConfig, Ipv6UnicastConfig, }; @@ -194,11 +193,7 @@ async fn run(args: RunArgs) { .expect("get BGP unnumbered neighbors from data store"), ); - start_bfd_sessions( - context.clone(), - db.get_bfd_neighbors() - .expect("get BFD neighbors from data store"), - ); + start_bfd_sessions(context.clone()); initialize_static_routes(&db, &context.log); @@ -476,15 +471,12 @@ fn start_bgp_routers( } } -fn start_bfd_sessions( - context: Arc, - configs: Vec, -) { - dlog!(context.log, info, "starting bfd sessions: {configs:#?}"); - for config in configs { - bfd_admin::add_peer(context.clone(), config) - .unwrap_or_else(|e| panic!("failed to add bfd peer {e}")); - } +fn start_bfd_sessions(context: Arc) { + dlog!(context.log, info, "starting persisted bfd sessions"); + context + .bfd + .restore_db_peers(context.db.clone()) + .unwrap_or_else(|e| panic!("failed to restore bfd peers: {e}")); } // Read static routes from disk, normalize prefixes by unsetting host bits, diff --git a/mgd/src/oxstats.rs b/mgd/src/oxstats.rs index 191ca35a..4c4c68cc 100644 --- a/mgd/src/oxstats.rs +++ b/mgd/src/oxstats.rs @@ -596,13 +596,7 @@ impl Stats { } fn bfd_stats(&mut self) -> Result, MetricsError> { - let daemon = lock!(self.bfd.daemon); - let mut counters = BTreeMap::new(); - for (addr, session) in daemon.sessions_iter() { - counters.insert(*addr, Arc::clone(session.counters())); - } - drop(daemon); - + let counters = self.bfd.session_counters(); let mut samples = Vec::with_capacity(counters.len() * 8); for (addr, counters) in &counters { diff --git a/rdb/src/db.rs b/rdb/src/db.rs index 819d1f23..61f3ad05 100644 --- a/rdb/src/db.rs +++ b/rdb/src/db.rs @@ -471,6 +471,16 @@ impl Db { let tree = self.persistent.open_tree(BFD_NEIGHBOR)?; let key = cfg.peer.to_string(); let value = serde_json::to_string(&cfg)?; + match tree.contains_key(key.as_str()) { + Ok(exists) => { + if exists { + return Err(Error::Conflict( + "bfd neighbor already exists".to_string(), + )); + } + } + Err(e) => return Err(Error::DataStore(e)), + } tree.insert(key.as_str(), value.as_str())?; tree.flush()?; Ok(()) @@ -533,7 +543,7 @@ impl Db { let current = self.get_origin4(asn)?; if !current.is_empty() { - return Err(Error::Conflict("origin already exists".to_string())); + return Err(Error::Conflict("origin4 already exists".to_string())); } self.set_origin4(asn, ps) @@ -600,7 +610,7 @@ impl Db { ) -> Result<(), Error> { let current = self.get_origin6(asn)?; if !current.is_empty() { - return Err(Error::Conflict("origin already exists".to_string())); + return Err(Error::Conflict("origin6 already exists".to_string())); } self.set_origin6(asn, ps)