110 lines
4.4 KiB
Rust
110 lines
4.4 KiB
Rust
//! The ping monitor as seen through the HTTP API. The socket is left out
|
|
//! (it needs privileges); observations are fed in directly.
|
|
|
|
use std::time::Duration;
|
|
|
|
use bgp_integration_tests::test_server::TestServer;
|
|
use bgp_primitives::ip_prefix::IpPrefix;
|
|
use bgp_rib::LocalRoute;
|
|
use bgp_server::ping::PingObservation;
|
|
use eyre::Result;
|
|
use serde_json::Value;
|
|
|
|
const TIMEOUT: Duration = Duration::from_secs(5);
|
|
|
|
fn ping(src: &str, dst: &str, hop_limit: u8, seq: u16) -> PingObservation {
|
|
PingObservation {
|
|
src: src.parse().unwrap(),
|
|
dst: dst.parse().unwrap(),
|
|
interface: Some("swissix0".into()),
|
|
hop_limit: Some(hop_limit),
|
|
ident: 1,
|
|
seq,
|
|
bytes: 64,
|
|
packet: [&[128, 0, 0xab, 0xcd, 0, 1][..], &seq.to_be_bytes(), b"hi"].concat(),
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn pings_are_listed_with_the_route_they_were_sent_to() -> Result<()> {
|
|
let server = TestServer::start(65000).await?;
|
|
let (base, _http) = server.serve_http().await?;
|
|
let client = reqwest::Client::new();
|
|
server
|
|
.local_rib
|
|
.announce(LocalRoute::new(IpPrefix::try_from("2001:db8:1a::/48").unwrap(), "alice", Default::default()))
|
|
.await?;
|
|
|
|
server.pings.record(ping("2001:67c:10ec::5", "2001:db8:1a::1", 57, 0)).await;
|
|
server.pings.record(ping("2001:67c:10ec::5", "2001:db8:1a::1", 57, 1)).await;
|
|
server.pings.record(ping("2a02:1210::9", "2001:db8:ff::1", 243, 0)).await;
|
|
server.pings.record(ping("2a02:1210::9", "2a0d:d740::1", 243, 0)).await; // not a watched prefix
|
|
|
|
let view: Value = client.get(format!("{base}/api/pings")).send().await?.error_for_status()?.json().await?;
|
|
assert_eq!(view["status"], "listening");
|
|
assert_eq!(view["prefixes"][0], "2001:db8::/32");
|
|
assert_eq!(view["total_flows"], 2);
|
|
|
|
let flows = view["flows"].as_array().unwrap();
|
|
let alice = flows.iter().find(|f| f["dst"] == "2001:db8:1a::1").unwrap();
|
|
assert_eq!(alice["src"], "2001:67c:10ec::5");
|
|
assert_eq!(alice["prefix"], "2001:db8:1a::/48");
|
|
assert_eq!(alice["owner"], "alice");
|
|
assert_eq!(alice["interface"], "swissix0");
|
|
assert_eq!((alice["hop_limit"].as_u64(), alice["hops"].as_u64()), (Some(57), Some(7)));
|
|
assert_eq!((alice["count"].as_u64(), alice["seq"].as_u64()), (Some(2), Some(1)));
|
|
assert_eq!(alice["hex"], "8000abcd000100016869");
|
|
assert!(alice["last_seen_unix"].as_u64().unwrap() > 1_700_000_000);
|
|
|
|
let stray = flows.iter().find(|f| f["dst"] == "2001:db8:ff::1").unwrap();
|
|
assert!(stray["prefix"].is_null() && stray["owner"].is_null());
|
|
assert_eq!(stray["hops"], 12);
|
|
|
|
let limited: Value = client.get(format!("{base}/api/pings?limit=1")).send().await?.json().await?;
|
|
assert_eq!(limited["total_flows"], 2);
|
|
assert_eq!(limited["flows"].as_array().unwrap().len(), 1);
|
|
|
|
let metrics = client.get(format!("{base}/metricz")).send().await?.text().await?;
|
|
assert!(metrics.contains("ping_monitor_echo_requests_total"));
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn pings_appear_on_the_event_stream() -> Result<()> {
|
|
let server = TestServer::start(65000).await?;
|
|
let (base, _http) = server.serve_http().await?;
|
|
let mut resp = reqwest::Client::new().get(format!("{base}/api/events")).send().await?.error_for_status()?;
|
|
|
|
// The subscription exists once the response has started; a ping recorded
|
|
// before that would be missed, so keep pinging until one shows up.
|
|
let pings = server.pings.clone();
|
|
let pinger = tokio::spawn(async move {
|
|
for seq in 0.. {
|
|
pings.record(ping("2001:67c:10ec::5", "2001:db8:1a::1", 57, seq)).await;
|
|
tokio::time::sleep(Duration::from_millis(50)).await;
|
|
}
|
|
});
|
|
|
|
let mut buf = String::new();
|
|
let event = tokio::time::timeout(TIMEOUT, async {
|
|
loop {
|
|
let chunk = resp.chunk().await?.ok_or_else(|| eyre::eyre!("stream ended"))?;
|
|
buf.push_str(std::str::from_utf8(&chunk)?);
|
|
for line in buf.lines().filter_map(|l| l.strip_prefix("data:")) {
|
|
let ev: Value = serde_json::from_str(line.trim())?;
|
|
if ev["type"] == "ping_seen" {
|
|
return Ok::<_, eyre::Error>(ev);
|
|
}
|
|
}
|
|
}
|
|
})
|
|
.await??;
|
|
pinger.abort();
|
|
|
|
assert_eq!(event["ping"]["src"], "2001:67c:10ec::5");
|
|
assert_eq!(event["ping"]["dst"], "2001:db8:1a::1");
|
|
assert_eq!(event["ping"]["interface"], "swissix0");
|
|
assert!(event["new_burst"].is_boolean());
|
|
Ok(())
|
|
}
|