diff --git a/acl-filter/src/context.rs b/acl-filter/src/context.rs index eed114ff29..bb80ba5795 100644 --- a/acl-filter/src/context.rs +++ b/acl-filter/src/context.rs @@ -10,8 +10,6 @@ use acl::dpdk::lookup::DpdkAclLookup; use acl::dpdk::rule::{AclFieldChunks, RuleSpec}; #[cfg(test)] use acl::reference::table::{RefRule, ReferenceTable}; -use concurrency::sync::LazyLock; -use concurrency::sync::atomic::{AtomicU64, Ordering}; use config::ConfigError; use config::external::overlay::ValidatedOverlay; use config::external::overlay::acl::{AclAction, AclProtoMatch, AclScope, ValidatedAclRule}; @@ -404,16 +402,41 @@ impl fmt::Debug for AnyTable { } } -// Lazily initialized so this compiles under the loom backend, whose AtomicU64::new is not const -// (each instance registers with the loom executor). The atomic itself is still the backend atomic, -// so fetch_add() stays instrumented; only construction is deferred. On every other backend LazyLock -// is a thin wrapper over an otherwise-const atomic. -static TABLE_SEQ: LazyLock = LazyLock::new(|| AtomicU64::new(0)); +concurrency::with_std! { + use concurrency::sync::LazyLock; + use concurrency::sync::atomic::{AtomicU64, Ordering}; + + static TABLE_SEQ: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + fn next_in_sequence() -> u64 { + TABLE_SEQ.fetch_add(1, Ordering::Relaxed) + } +} + +concurrency::with_loom! { + // nosemgrep: rust-no-direct-std-sync-import + static TABLE_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + + fn next_in_sequence() -> u64 { + // nosemgrep: rust-no-direct-std-sync-import + TABLE_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + } +} + +concurrency::with_shuttle! { + // nosemgrep: rust-no-direct-std-sync-import + static TABLE_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + + fn next_in_sequence() -> u64 { + // nosemgrep: rust-no-direct-std-sync-import + TABLE_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + } +} /// A process-unique rte_acl context name. rte_acl rejects duplicate names, and a hot-swap briefly /// keeps the old and new contexts alive at once, so the name must be unique across the process. fn table_name(base: &str) -> String { - format!("acl_{base}_{}", TABLE_SEQ.fetch_add(1, Ordering::Relaxed)) + format!("acl_{base}_{}", next_in_sequence()) } /// Build one table for the selected backend from rules in precedence (insertion) order. diff --git a/dataplane/src/drivers/kernel/worker.rs b/dataplane/src/drivers/kernel/worker.rs index 5e3f41b095..c78fb63d76 100644 --- a/dataplane/src/drivers/kernel/worker.rs +++ b/dataplane/src/drivers/kernel/worker.rs @@ -204,7 +204,7 @@ impl Worker { // awaits before reading anything from the socket. _ = ticker.tick() => { intf.watchdog.pat(); - continue; + Vec::new() } }; @@ -221,11 +221,6 @@ impl Worker { let mut tx_drops: u64 = 0; // number of packets dropped on tx let rx_pkts = packets_vec.len() as u64; // number of packets received counters.rx = rx_pkts; - if rx_pkts == 0 { - // nothing to process, but the read may have hit errors worth reporting - intf.watchdog.record(&counters); - continue; - } let packets = packets_vec.into_iter(); let out_pkts = pipeline diff --git a/dataplane/src/packet_processor/fuzz.rs b/dataplane/src/packet_processor/fuzz.rs index 095c91c641..e158de9fcf 100644 --- a/dataplane/src/packet_processor/fuzz.rs +++ b/dataplane/src/packet_processor/fuzz.rs @@ -7,6 +7,7 @@ use acl_filter::{ AclFilter, AclFilterContext, AclFilterContextReaderFactory, AclFilterContextWriter, }; +use common::cliprovider::CliDataProvider; use concurrency::sync::{Arc, Mutex}; use config::external::GenId; use config::external::overlay::acl::Acl; @@ -222,7 +223,46 @@ impl Fleet { } } +pub(crate) struct CliReaders { + sources: Vec<(&'static str, Box)>, +} + +impl CliReaders { + pub(crate) fn read_one(&self, which: usize) -> (&'static str, String) { + let (name, source) = &self.sources[which % self.sources.len()]; + (name, source.provide()) + } + + pub(crate) fn len(&self) -> usize { + self.sources.len() + } +} + impl Blueprint { + pub(crate) fn cli_readers(&self) -> CliReaders { + CliReaders { + sources: vec![ + ("show flow-table", Box::new(self.flow_table.clone())), + ( + "show flow-filter", + Box::new(self.flow_filter.handle().inner()), + ), + ( + "show port-forwarding", + Box::new(self.portfw.handle().inner()), + ), + ( + "show static-nat", + Box::new(self.static_nat.handle().inner()), + ), + ( + "show masquerading", + Box::new(self.masquerade.handle().inner()), + ), + ], + } + } + pub(crate) fn worker(&self) -> Worker { let translations = Arc::new(Mutex::new(Translations::declaring(&self.declared))); let mut pipeline = DynPipeline::new().set_data(self.pipeline.clone()); @@ -750,7 +790,7 @@ pub(crate) mod derive { use super::routed::{Blast, Conversation, Inbound}; use super::*; use config::external::overlay::ValidatedOverlay; - use config::external::overlay::algebra::{Draft, Guard}; + use config::external::overlay::algebra::Draft; use config::external::overlay::vpcpeering::ValidatedExpose; use lpm::prefix::with_ports::L4Protocol; use lpm::prefix::{Prefix, PrefixPortsSet, PrefixWithOptionalPorts}; @@ -2068,7 +2108,7 @@ mod acl { ); } - fn prefix(text: &str) -> Prefix { + pub(super) fn prefix(text: &str) -> Prefix { text.parse() .unwrap_or_else(|_| unreachable!("a well-formed prefix")) } @@ -4569,7 +4609,13 @@ mod model { #[concurrency::model_test] fn two_workers_are_not_given_the_same_public_tuple() { - const FLOWS: u16 = 3; + const CASES: usize = 64; + + static ALLOCATED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static BARE: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static UNBUILT: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + use lpm::prefix::with_ports::PrefixWithOptionalPorts; let _eal = dpdk::test_support::start_eal(); @@ -4584,85 +4630,140 @@ mod model { }; let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); - concurrency::stress(move || { - let overlay = overlay_with_exposes_and_acl(exposes(), None) - .expect("the fixture exposes form an overlay") - .validate() - .expect("the fixture overlay validates"); - let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); - let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); - let blueprint = fleet.blueprint(); - - let workers = [("1.1.0.1", 1000u16), ("1.1.0.2", 2000u16)]; - let given: Vec = thread::scope(|scope| { - let running: Vec<_> = workers - .iter() - .map(|(host, first)| { - let entering = handle.clone(); - thread::Builder::new() - .name(format!("worker-{host}")) - .spawn_scoped(scope, move || { - let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); - let recording = - tracectl::evidence::capture(format!("worker-{host}")); - let mut worker = blueprint.worker(); - let burst = (0..FLOWS) - .map(|n| { - tunnelled(&build_test_udp_ipv4_packet( - host, - "3.3.3.1", - first + n, - 80, - )) - }) - .collect(); - let tuples = worker - .send_batch(burst) - .iter() - .map(|out| { - assert!( - matches!(verdict(out), Verdict::Delivered { .. }), - "a packet that is delivered single-threaded was not: \ - {:?}", - verdict(out) - ); - let tenant = inside(out).expect( - "a delivered frame leaves this gateway tunnelled", - ); - ( - tenant.ip_source(), - tenant.transport_src_port().map(std::num::NonZero::get), - ) + bolero::check!() + .with_max_len(MAX_INPUT_LEN) + .with_type() + .cloned() + .with_iterations(CASES) + .for_each(|(flows, addr_bits, first_a, first_b): (u8, u8, u16, u16)| { + let flows = u16::from(flows % 6) + 1; + let addr_bits = addr_bits % 3; + let (first_a, first_b) = (first_a % 20000 + 1000, first_b % 20000 + 30000); + + let Ok(public) = format!("2.2.0.0/{}", 32 - u32::from(addr_bits)).parse::() + else { + UNBUILT.fetch_add(1, Ordering::Relaxed); + return; + }; + let drawn = VpcExpose::empty() + .make_masquerade(None) + .expect("masquerade is a legal flavour for an empty expose") + .ip("1.1.0.0/16" + .parse::() + .expect("a literal prefix") + .into()) + .as_range(PrefixWithOptionalPorts::new(public, None)); + let Ok(drawn) = drawn else { + UNBUILT.fetch_add(1, Ordering::Relaxed); + return; + }; + let overlay = overlay_with_exposes_and_acl(vec![drawn], None) + .and_then(|overlay| overlay.validate()); + let Ok(overlay) = overlay else { + UNBUILT.fetch_add(1, Ordering::Relaxed); + return; + }; + + let entering = handle.clone(); + concurrency::stress(move || { + let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + let fleet = + Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + + let workers = [("1.1.0.1", first_a), ("1.1.0.2", first_b)]; + let given: Vec = thread::scope(|scope| { + let running: Vec<_> = workers + .iter() + .map(|(host, first)| { + let entering = entering.clone(); + thread::Builder::new() + .name(format!("worker-{host}")) + .spawn_scoped(scope, move || { + let _guard = + entering.as_ref().map(tokio::runtime::Handle::enter); + let recording = + tracectl::evidence::capture(format!("worker-{host}")); + let mut worker = blueprint.worker(); + let burst = (0..flows) + .map(|n| { + tunnelled(&build_test_udp_ipv4_packet( + host, + "3.3.3.1", + first + n, + 80, + )) + }) + .collect(); + let tuples = worker + .send_batch(burst) + .iter() + .filter(|out| { + matches!(verdict(out), Verdict::Delivered { .. }) + }) + .map(|out| { + let tenant = inside(out).expect( + "a delivered frame leaves this gateway \ + tunnelled", + ); + ( + tenant.ip_source(), + tenant + .transport_src_port() + .map(std::num::NonZero::get), + ) + }) + .collect::>(); + (tuples, recording.evidence()) }) - .collect::>(); - (tuples, recording.evidence()) + .expect("spawn worker") }) - .expect("spawn worker") - }) - .collect(); - running - .into_iter() - .map(|worker| worker.join().expect("worker panicked")) - .collect() - }); + .collect(); + running + .into_iter() + .map(|worker| worker.join().expect("worker panicked")) + .collect() + }); - let (given, evidence): (Vec>, Vec<_>) = given.into_iter().unzip(); - let given: Vec<_> = given.into_iter().flatten().collect(); - let _explain = tracectl::evidence::dump_on_panic(evidence); + let (given, evidence): (Vec>, Vec<_>) = given.into_iter().unzip(); + let given: Vec<_> = given.into_iter().flatten().collect(); + let _explain = tracectl::evidence::dump_on_panic(evidence); - let mut seen = std::collections::BTreeMap::new(); - for tuple in &given { - let count: &mut usize = seen.entry(format!("{tuple:?}")).or_default(); - *count += 1; - assert_eq!( - *count, - 1, - "{} distinct flows were translated and {tuple:?} was handed out twice, so a \ - reply to it cannot be attributed to either of them: {given:?}", - given.len() - ); - } - }); + if given.is_empty() { + BARE.fetch_add(1, Ordering::Relaxed); + } else { + ALLOCATED.fetch_add(1, Ordering::Relaxed); + } + + let mut seen = std::collections::BTreeMap::new(); + for tuple in &given { + let count: &mut usize = seen.entry(format!("{tuple:?}")).or_default(); + *count += 1; + assert_eq!( + *count, + 1, + "{} distinct flows were translated and {tuple:?} was handed out \ + twice, so a reply to it cannot be attributed to either of them: \ + {given:?}", + given.len() + ); + } + }); + }); + + let (allocated, bare, unbuilt) = ( + ALLOCATED.load(Ordering::Relaxed), + BARE.load(Ordering::Relaxed), + UNBUILT.load(Ordering::Relaxed), + ); + eprintln!("allocated={allocated} bare={bare} unbuilt={unbuilt}"); + super::assert_covered( + allocated > 0, + &format!( + "no draw translated a single flow, so nothing was ever compared \ + (allocated={allocated} bare={bare} unbuilt={unbuilt})" + ), + ); } #[concurrency::model_test] @@ -4791,7 +4892,7 @@ mod model { #[concurrency::model_test] fn a_reply_is_translated_by_a_worker_that_never_saw_the_request() { - const FLOWS: u8 = 2; + const CASES: usize = 64; static CLOSED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static ABANDONED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); @@ -4809,98 +4910,117 @@ mod model { }; let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); - concurrency::stress(move || { - let overlay = overlay_with_exposes_and_acl(exposes(), None) - .expect("the fixture exposes form an overlay") - .validate() - .expect("the fixture overlay validates"); - let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); - let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); - let blueprint = fleet.blueprint(); - let entering = handle.clone(); + bolero::check!() + .with_max_len(MAX_INPUT_LEN) + .with_type() + .cloned() + .with_iterations(CASES) + .for_each(|(flows, hosts): (u8, u8)| { + let flows = flows % 4 + 1; + let hosts = hosts % 8; + let entering = handle.clone(); + concurrency::stress(move || { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + let fleet = + Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); - let mut opened: Vec> = thread::scope(|scope| { - let running: Vec<_> = (0..2u8) - .map(|which| { - let entering = entering.clone(); - thread::Builder::new() - .name(format!("open-{which}")) - .spawn_scoped(scope, move || { - let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); - let _evidence = - tracectl::evidence::capture(format!("open-{which}")); - let mut worker = blueprint.worker(); - (0..FLOWS) - .map(|nth| { - let src = format!("1.1.{which}.{}", nth + 1); - let mut convo = Conversation::new( - super::routed::Path::fixture(), - src.parse().unwrap_or_else(|e| { - unreachable!("{src} is an address: {e}") - }), - "3.3.3.1" - .parse() - .unwrap_or_else(|e| unreachable!("{e}")), - u16::from(nth) + 1000, - 80, - ); - step(&mut worker, &mut convo); - convo + let mut opened: Vec> = thread::scope(|scope| { + let running: Vec<_> = (0..2u8) + .map(|which| { + let entering = entering.clone(); + thread::Builder::new() + .name(format!("open-{which}")) + .spawn_scoped(scope, move || { + let _guard = + entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("open-{which}")); + let mut worker = blueprint.worker(); + (0..flows) + .map(|nth| { + let src = + format!("1.1.{which}.{}", hosts + nth + 1); + let mut convo = Conversation::new( + super::routed::Path::fixture(), + src.parse().unwrap_or_else(|e| { + unreachable!("{src} is an address: {e}") + }), + "3.3.3.1" + .parse() + .unwrap_or_else(|e| unreachable!("{e}")), + u16::from(nth) + 1000, + 80, + ); + step(&mut worker, &mut convo); + convo + }) + .collect::>() }) - .collect::>() + .expect("spawn opener") }) - .expect("spawn opener") - }) - .collect(); - running - .into_iter() - .map(|opener| opener.join().expect("opener panicked")) - .collect() - }); + .collect(); + running + .into_iter() + .map(|opener| opener.join().expect("opener panicked")) + .collect() + }); - let second = opened - .pop() - .unwrap_or_else(|| unreachable!("two openers ran")); - let first = opened - .pop() - .unwrap_or_else(|| unreachable!("two openers ran")); - - let answered: Vec> = thread::scope(|scope| { - let running: Vec<_> = [(0u8, second), (1u8, first)] - .into_iter() - .map(|(which, mut theirs)| { - let entering = entering.clone(); - thread::Builder::new() - .name(format!("answer-{which}")) - .spawn_scoped(scope, move || { - let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); - let _evidence = - tracectl::evidence::capture(format!("answer-{which}")); - let mut worker = blueprint.worker(); - for convo in &mut theirs { - if !convo.finished() { - step(&mut worker, convo); - } - } - theirs + let second = opened + .pop() + .unwrap_or_else(|| unreachable!("two openers ran")); + let first = opened + .pop() + .unwrap_or_else(|| unreachable!("two openers ran")); + + let answered: Vec> = thread::scope(|scope| { + let running: Vec<_> = [(0u8, second), (1u8, first)] + .into_iter() + .map(|(which, mut theirs)| { + let entering = entering.clone(); + thread::Builder::new() + .name(format!("answer-{which}")) + .spawn_scoped(scope, move || { + let _guard = + entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("answer-{which}")); + let mut worker = blueprint.worker(); + for convo in &mut theirs { + if !convo.finished() { + step(&mut worker, convo); + } + } + theirs + }) + .expect("spawn answerer") }) - .expect("spawn answerer") - }) - .collect(); - running - .into_iter() - .map(|answerer| answerer.join().expect("answerer panicked")) - .collect() + .collect(); + running + .into_iter() + .map(|answerer| answerer.join().expect("answerer panicked")) + .collect() + }); + + for convo in answered.into_iter().flatten() { + if convo.checked() { + CLOSED.fetch_add(1, Ordering::Relaxed); + } else { + ABANDONED.fetch_add(1, Ordering::Relaxed); + } + } + }); }); - for convo in answered.into_iter().flatten() { - if convo.checked() { - CLOSED.fetch_add(1, Ordering::Relaxed); - } else { - ABANDONED.fetch_add(1, Ordering::Relaxed); - } - } - }); + let (closed, abandoned) = ( + CLOSED.load(Ordering::Relaxed), + ABANDONED.load(Ordering::Relaxed), + ); + eprintln!("closed={closed} abandoned={abandoned}"); let (closed, abandoned) = ( CLOSED.load(Ordering::Relaxed), @@ -4940,6 +5060,8 @@ mod model { #[concurrency::model_test] fn an_icmp_teardown_leaves_another_workers_flow_alone() { + const CASES: usize = 64; + static REPORTED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static SURVIVED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); @@ -4956,141 +5078,154 @@ mod model { }; let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); - concurrency::stress(move || { - let target: IpAddr = "3.3.3.1".parse().unwrap_or_else(|e| unreachable!("{e}")); + bolero::check!() + .with_max_len(MAX_INPUT_LEN) + .with_type() + .cloned() + .with_iterations(CASES) + .for_each(|(spare, doomed_port, spared_port): (u8, u16, u16)| { + let spare = spare % 8 + 2; + let doomed_port = doomed_port % 20000 + 1000; + let spared_port = spared_port % 20000 + 30000; + let handle = handle.clone(); + concurrency::stress(move || { + let target: IpAddr = "3.3.3.1".parse().unwrap_or_else(|e| unreachable!("{e}")); - for (code, tears_down) in [ - (net::icmp4::Icmp4DestUnreachable::Network, true), - ( - net::icmp4::Icmp4DestUnreachable::FragmentationNeeded { - next_hop_mtu: Some(1400.try_into().unwrap_or_else(|_| unreachable!())), - }, - false, - ), - ] { - let overlay = overlay_with_exposes_and_acl(exposes(), None) - .expect("the fixture exposes form an overlay") - .validate() - .expect("the fixture overlay validates"); - let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); - let fleet = - Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); - let blueprint = fleet.blueprint(); - let entering = handle.clone(); + for (code, tears_down) in [ + (net::icmp4::Icmp4DestUnreachable::Network, true), + ( + net::icmp4::Icmp4DestUnreachable::FragmentationNeeded { + next_hop_mtu: Some(1400.try_into().unwrap_or_else(|_| unreachable!())), + }, + false, + ), + ] { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + let fleet = + Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let entering = handle.clone(); - let opening = entering.clone(); - let answering = entering.clone(); - let (doomed, mut spared): ((IpAddr, u16), Conversation) = thread::scope(|scope| { - let doomed = thread::Builder::new() - .name("open-doomed".to_owned()) - .spawn_scoped(scope, move || { - let _guard = opening.as_ref().map(tokio::runtime::Handle::enter); - let _evidence = tracectl::evidence::capture("open-doomed"); - let mut worker = blueprint.worker(); - let request = super::round_trip::udp( - "1.1.0.1".parse().unwrap_or_else(|e| unreachable!("{e}")), - target, - 1000, - 80, - ) - .unwrap_or_else(|| unreachable!("the fixture request builds")); - let out = worker.send(tunnelled(&request)); - let carried = inside(&out).unwrap_or_else(|| { - unreachable!( - "the request was not delivered tunnelled: {:?}", - verdict(&out) + let opening = entering.clone(); + let answering = entering.clone(); + let (doomed, mut spared): ((IpAddr, u16), Conversation) = thread::scope(|scope| { + let doomed = thread::Builder::new() + .name("open-doomed".to_owned()) + .spawn_scoped(scope, move || { + let _guard = opening.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = tracectl::evidence::capture("open-doomed"); + let mut worker = blueprint.worker(); + let request = super::round_trip::udp( + "1.1.0.1".parse().unwrap_or_else(|e| unreachable!("{e}")), + target, + doomed_port, + 80, ) - }); - let (Some(public), Some(port)) = - (carried.ip_source(), carried.transport_src_port()) - else { - unreachable!("a delivered request had no public tuple") - }; - (public, port.get()) - }) - .expect("spawn opener"); + .unwrap_or_else(|| unreachable!("the fixture request builds")); + let out = worker.send(tunnelled(&request)); + let carried = inside(&out).unwrap_or_else(|| { + unreachable!( + "the request was not delivered tunnelled: {:?}", + verdict(&out) + ) + }); + let (Some(public), Some(port)) = + (carried.ip_source(), carried.transport_src_port()) + else { + unreachable!("a delivered request had no public tuple") + }; + (public, port.get()) + }) + .expect("spawn opener"); - let spared = thread::Builder::new() - .name("open-spared".to_owned()) - .spawn_scoped(scope, move || { - let _guard = answering.as_ref().map(tokio::runtime::Handle::enter); - let _evidence = tracectl::evidence::capture("open-spared"); - let mut worker = blueprint.worker(); - let mut convo = Conversation::new( - super::routed::Path::fixture(), - "1.1.0.2".parse().unwrap_or_else(|e| unreachable!("{e}")), - target, - 2000, - 80, - ); - step(&mut worker, &mut convo); - convo - }) - .expect("spawn opener"); + let spared = thread::Builder::new() + .name("open-spared".to_owned()) + .spawn_scoped(scope, move || { + let _guard = answering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = tracectl::evidence::capture("open-spared"); + let mut worker = blueprint.worker(); + let mut convo = Conversation::new( + super::routed::Path::fixture(), + format!("1.1.0.{spare}") + .parse() + .unwrap_or_else(|e| unreachable!("{e}")), + target, + spared_port, + 80, + ); + step(&mut worker, &mut convo); + convo + }) + .expect("spawn opener"); - ( - doomed.join().expect("opener panicked"), - spared.join().expect("opener panicked"), - ) - }); + ( + doomed.join().expect("opener panicked"), + spared.join().expect("opener panicked"), + ) + }); - let tearing = entering.clone(); - let keeping = entering.clone(); - let named = format!("{code:?}"); - let spared = thread::scope(|scope| { - let teardown = thread::Builder::new() - .name("teardown".to_owned()) - .spawn_scoped(scope, move || { - let _guard = tearing.as_ref().map(tokio::runtime::Handle::enter); - let _evidence = tracectl::evidence::capture("teardown"); - let mut worker = blueprint.worker(); - let out = worker.send(unreachable(code, doomed, target)); - matches!(verdict(&out), Verdict::Delivered { .. }) - }) - .expect("spawn teardown"); + let tearing = entering.clone(); + let keeping = entering.clone(); + let named = format!("{code:?}"); + let spared = thread::scope(|scope| { + let teardown = thread::Builder::new() + .name("teardown".to_owned()) + .spawn_scoped(scope, move || { + let _guard = tearing.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = tracectl::evidence::capture("teardown"); + let mut worker = blueprint.worker(); + let out = worker.send(unreachable(code, doomed, target)); + matches!(verdict(&out), Verdict::Delivered { .. }) + }) + .expect("spawn teardown"); - let answer = thread::Builder::new() - .name("answer".to_owned()) - .spawn_scoped(scope, move || { - let _guard = keeping.as_ref().map(tokio::runtime::Handle::enter); - let _evidence = tracectl::evidence::capture("answer"); - let mut worker = blueprint.worker(); - step(&mut worker, &mut spared); - spared - }) - .expect("spawn answer"); + let answer = thread::Builder::new() + .name("answer".to_owned()) + .spawn_scoped(scope, move || { + let _guard = keeping.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = tracectl::evidence::capture("answer"); + let mut worker = blueprint.worker(); + step(&mut worker, &mut spared); + spared + }) + .expect("spawn answer"); + + assert!( + teardown.join().expect("teardown panicked"), + "the icmp error never reached the flow it named, so this raced against \ + nothing" + ); + REPORTED.fetch_add(1, Ordering::Relaxed); + answer.join().expect("answer panicked") + }); assert!( - teardown.join().expect("teardown panicked"), - "the icmp error never reached the flow it named, so this raced against \ - nothing" + spared.checked(), + "a flow was disturbed by an icmp teardown of a different flow on another \ + worker. {}", + spared.describe() ); - REPORTED.fetch_add(1, Ordering::Relaxed); - answer.join().expect("answer panicked") - }); - - assert!( - spared.checked(), - "a flow was disturbed by an icmp teardown of a different flow on another \ - worker. {}", - spared.describe() - ); - SURVIVED.fetch_add(1, Ordering::Relaxed); + SURVIVED.fetch_add(1, Ordering::Relaxed); - let mut worker = blueprint.worker(); - let reply = super::round_trip::udp(target, doomed.0, 80, doomed.1) - .unwrap_or_else(|| unreachable!("the reply builds")); - let out = worker.send(super::routed::tunnelled_from(vni(REMOTE_VNI), &reply)); - let delivered = matches!(verdict(&out), Verdict::Delivered { .. }); - assert_eq!( - delivered, - !tears_down, - "an icmp error with code {named} left the flow it named {}: {:?}", - if delivered { "alive" } else { "torn down" }, - verdict(&out) - ); - } - }); + let mut worker = blueprint.worker(); + let reply = super::round_trip::udp(target, doomed.0, 80, doomed.1) + .unwrap_or_else(|| unreachable!("the reply builds")); + let out = worker.send(super::routed::tunnelled_from(vni(REMOTE_VNI), &reply)); + let delivered = matches!(verdict(&out), Verdict::Delivered { .. }); + assert_eq!( + delivered, + !tears_down, + "an icmp error with code {named} left the flow it named {}: {:?}", + if delivered { "alive" } else { "torn down" }, + verdict(&out) + ); + } + }); + }); let (reported, survived) = ( REPORTED.load(Ordering::Relaxed), @@ -5102,7 +5237,7 @@ mod model { #[concurrency::model_test] fn forwarding_survives_a_route_being_republished_underneath_it() { - const FLOWS: u8 = 2; + const CASES: usize = 64; const CHURN: u8 = 6; static COMPLETED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); @@ -5121,83 +5256,94 @@ mod model { }; let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); - concurrency::stress(move || { - let overlay = overlay_with_exposes_and_acl(exposes(), None) - .expect("the fixture exposes form an overlay") - .validate() - .expect("the fixture overlay validates"); - let mut tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); - let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); - let blueprint = fleet.blueprint(); - let entering = handle.clone(); - let start = Arc::new(concurrency::sync::Barrier::new(3)); - - thread::scope(|scope| { - let running: Vec<_> = (0..2u8) - .map(|which| { - let entering = entering.clone(); - let start = start.clone(); - thread::Builder::new() - .name(format!("forward-{which}")) - .spawn_scoped(scope, move || { - let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); - let _evidence = - tracectl::evidence::capture(format!("forward-{which}")); - let mut worker = blueprint.worker(); - start.wait(); - let mut done = 0u64; - for nth in 0..FLOWS { - let src = format!("1.1.{which}.{}", nth + 1); - let mut convo = Conversation::new( - super::routed::Path::fixture(), - src.parse().unwrap_or_else(|e| unreachable!("{src}: {e}")), - "3.3.3.1".parse().unwrap_or_else(|e| unreachable!("{e}")), - u16::from(nth) + 1000, - 80, - ); - drive(&mut worker, &mut convo); - assert!( - convo.checked(), - "a conversation did not complete while routes were being \ - republished. {}", - convo.describe() - ); - done += 1; - } - done - }) - .expect("spawn forwarder") - }) - .collect(); + bolero::check!() + .with_max_len(MAX_INPUT_LEN) + .with_type() + .cloned() + .with_iterations(CASES) + .for_each(|(flows, churn, host): (u8, u8, u8)| { + let flows = flows % 4 + 1; + let churn = churn % 8 + 1; + let host = host % 8; + let handle = handle.clone(); + concurrency::stress(move || { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let mut tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let entering = handle.clone(); + let start = Arc::new(concurrency::sync::Barrier::new(3)); + + thread::scope(|scope| { + let running: Vec<_> = (0..2u8) + .map(|which| { + let entering = entering.clone(); + let start = start.clone(); + thread::Builder::new() + .name(format!("forward-{which}")) + .spawn_scoped(scope, move || { + let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("forward-{which}")); + let mut worker = blueprint.worker(); + start.wait(); + let mut done = 0u64; + for nth in 0..flows { + let src = format!("1.1.{which}.{}", host + nth + 1); + let mut convo = Conversation::new( + super::routed::Path::fixture(), + src.parse().unwrap_or_else(|e| unreachable!("{src}: {e}")), + "3.3.3.1".parse().unwrap_or_else(|e| unreachable!("{e}")), + u16::from(nth) + 1000, + 80, + ); + drive(&mut worker, &mut convo); + assert!( + convo.checked(), + "a conversation did not complete while routes were being \ + republished. {}", + convo.describe() + ); + done += 1; + } + done + }) + .expect("spawn forwarder") + }) + .collect(); - let peer: IpAddr = PEER_VTEP.parse().unwrap_or_else(|_| unreachable!()); - let landing = - FibGroup::with_entry(FibEntry::with_inst(PktInstruction::Local(uplink()))); - start.wait(); - for nth in 0..CHURN { - let prefix = format!("9.9.{nth}.0"); - tables.route_via( - UNDERLAY_VRF, - Prefix::expect_from(( - prefix - .parse::() - .unwrap_or_else(|e| unreachable!("{e}")), - 24, - )), - nhop(&peer), - &landing, - ); - PUBLISHED.fetch_add(1, Ordering::Relaxed); - } + let peer: IpAddr = PEER_VTEP.parse().unwrap_or_else(|_| unreachable!()); + let landing = + FibGroup::with_entry(FibEntry::with_inst(PktInstruction::Local(uplink()))); + start.wait(); + for nth in 0..churn { + let prefix = format!("9.9.{nth}.0"); + tables.route_via( + UNDERLAY_VRF, + Prefix::expect_from(( + prefix + .parse::() + .unwrap_or_else(|e| unreachable!("{e}")), + 24, + )), + nhop(&peer), + &landing, + ); + PUBLISHED.fetch_add(1, Ordering::Relaxed); + } - for worker in running { - COMPLETED.fetch_add( - worker.join().expect("forwarder panicked"), - Ordering::Relaxed, - ); - } + for worker in running { + COMPLETED.fetch_add( + worker.join().expect("forwarder panicked"), + Ordering::Relaxed, + ); + } + }); + }); }); - }); let (completed, published) = ( COMPLETED.load(Ordering::Relaxed), @@ -5212,8 +5358,7 @@ mod model { #[concurrency::model_test] fn a_next_hop_that_moves_is_never_seen_half_moved() { - const CHURN: u8 = 3; - const PER_ROUND: u8 = 2; + const CASES: usize = 64; static FRESH: LazyLock = LazyLock::new(|| AtomicU64::new(0)); static STALE: LazyLock = LazyLock::new(|| AtomicU64::new(0)); @@ -5268,153 +5413,164 @@ mod model { }; let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); - concurrency::stress(move || { - let overlay = overlay_with_exposes_and_acl(exposes(), None) - .expect("the fixture exposes form an overlay") - .validate() - .expect("the fixture overlay validates"); - let mut tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); - for nth in 1..=CHURN { - let (_, oif) = waypoint(nth); - tables.interface( - oif, - &format!("uplink-{nth}"), - SourceMac::new(framing(nth)).unwrap_or_else(|_| unreachable!()), - ); - tables.attach(oif, UNDERLAY_VRF); - } - for to in 0..=CHURN { - for over in 0..=CHURN { - tables.adjacency(waypoint(to).0, waypoint(over).1, framing(to)); + bolero::check!() + .with_max_len(MAX_INPUT_LEN) + .with_type() + .cloned() + .with_iterations(CASES) + .for_each(|(churn, per_round, host): (u8, u8, u8)| { + let churn = churn % 4 + 1; + let per_round = per_round % 3 + 1; + let host = host % 8 + 1; + let handle = handle.clone(); + concurrency::stress(move || { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let mut tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + for nth in 1..=churn { + let (_, oif) = waypoint(nth); + tables.interface( + oif, + &format!("uplink-{nth}"), + SourceMac::new(framing(nth)).unwrap_or_else(|_| unreachable!()), + ); + tables.attach(oif, UNDERLAY_VRF); } - } - let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); - let blueprint = fleet.blueprint(); - let entering = handle.clone(); - let gate = Arc::new(concurrency::sync::Barrier::new(3)); - let mut reports = Vec::new(); - - thread::scope(|scope| { - let running: Vec<_> = (0..2u8) - .map(|which| { - let entering = entering.clone(); - let gate = gate.clone(); - thread::Builder::new() - .name(format!("probe-{which}")) - .spawn_scoped(scope, move || { - let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); - let _evidence = - tracectl::evidence::capture(format!("probe-{which}")); - let mut worker = blueprint.worker(); - - let mut port = u16::from(which) * 1000 + 1000; - let send = |worker: &mut Worker, port: &mut u16| -> Seen { - *port += 1; - let src = format!("1.1.{which}.1"); - let out = worker.send(tunnelled(&build_test_udp_ipv4_packet( - &src, "3.3.3.1", *port, 80, - ))); - let Verdict::Delivered { - oif: Some(oif), - dst: Some(dst), - .. - } = verdict(&out) - else { - return Err(format!( - "it did not leave the gateway: {:?}", - verdict(&out) - )); + for to in 0..=churn { + for over in 0..=churn { + tables.adjacency(waypoint(to).0, waypoint(over).1, framing(to)); + } + } + let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let entering = handle.clone(); + let gate = Arc::new(concurrency::sync::Barrier::new(3)); + let mut reports = Vec::new(); + + thread::scope(|scope| { + let running: Vec<_> = (0..2u8) + .map(|which| { + let entering = entering.clone(); + let gate = gate.clone(); + thread::Builder::new() + .name(format!("probe-{which}")) + .spawn_scoped(scope, move || { + let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("probe-{which}")); + let mut worker = blueprint.worker(); + + let mut port = u16::from(which) * 1000 + 1000; + let send = |worker: &mut Worker, port: &mut u16| -> Seen { + *port += 1; + let src = format!("1.1.{which}.{host}"); + let out = worker.send(tunnelled(&build_test_udp_ipv4_packet( + &src, "3.3.3.1", *port, 80, + ))); + let Verdict::Delivered { + oif: Some(oif), + dst: Some(dst), + .. + } = verdict(&out) + else { + return Err(format!( + "it did not leave the gateway: {:?}", + verdict(&out) + )); + }; + (0..=churn) + .find(|nth| waypoint(*nth) == (dst, oif)) + .ok_or_else(|| { + format!( + "it left over interface {oif} towards {dst}, which \ + is no published next hop: either the \ + encapsulation and the egress came from different \ + versions, or the group was read while it was \ + being written" + ) + }) }; - (0..=CHURN) - .find(|nth| waypoint(*nth) == (dst, oif)) - .ok_or_else(|| { - format!( - "it left over interface {oif} towards {dst}, which \ - is no published next hop: either the \ - encapsulation and the egress came from different \ - versions, or the group was read while it was \ - being written" - ) - }) - }; - let probe = |worker: &mut Worker, port: &mut u16| -> Seen { - without_unwinding(|| send(worker, port)).unwrap_or_else(|why| { - Err(format!("sending it panicked: {why}")) - }) - }; + let probe = |worker: &mut Worker, port: &mut u16| -> Seen { + without_unwinding(|| send(worker, port)).unwrap_or_else(|why| { + Err(format!("sending it panicked: {why}")) + }) + }; - gate.wait(); - let mut seen = Vec::with_capacity( - usize::from(CHURN) * usize::from(PER_ROUND) + 1, - ); - for _ in 1..=CHURN { - for _ in 0..PER_ROUND { - seen.push(probe(&mut worker, &mut port)); - } gate.wait(); - } - seen.push(probe(&mut worker, &mut port)); - seen - }) - .expect("spawn prober") - }) - .collect(); + let mut seen = Vec::with_capacity( + usize::from(churn) * usize::from(per_round) + 1, + ); + for _ in 1..=churn { + for _ in 0..per_round { + seen.push(probe(&mut worker, &mut port)); + } + gate.wait(); + } + seen.push(probe(&mut worker, &mut port)); + seen + }) + .expect("spawn prober") + }) + .collect(); - let key = nhop(&PEER_VTEP.parse().unwrap_or_else(|_| unreachable!())); - gate.wait(); - for round in 1..=CHURN { - tables.nexthop(REMOTE_VNI, &key, &towards(round)); + let key = nhop(&PEER_VTEP.parse().unwrap_or_else(|_| unreachable!())); gate.wait(); - } + for round in 1..=churn { + tables.nexthop(REMOTE_VNI, &key, &towards(round)); + gate.wait(); + } - for prober in running { - reports.push(prober.join().expect("prober panicked")); - } - }); + for prober in running { + reports.push(prober.join().expect("prober panicked")); + } + }); - for seen in reports { - for (nth, observed) in seen.iter().enumerate() { - let last = nth == seen.len() - 1; - let round = if last { - CHURN - } else { - u8::try_from(nth).unwrap_or_else(|_| unreachable!()) / PER_ROUND + 1 - }; - let version = match observed { - Ok(version) => *version, - Err(why) => panic!( - "a probe sent in round {round}, while the next hop was moving, is not \ - attributable to any version: {why}" - ), - }; - if last { - assert_eq!( - version, CHURN, - "a probe sent after the churn had finished was forwarded by version \ - {version}, not by version {CHURN}, the last one published. Every \ - publish returned before the barrier that released this probe, so a \ - reader still serving an earlier version is serving a next hop that \ - no longer exists" + for seen in reports { + for (nth, observed) in seen.iter().enumerate() { + let last = nth == seen.len() - 1; + let round = if last { + churn + } else { + u8::try_from(nth).unwrap_or_else(|_| unreachable!()) / per_round + 1 + }; + let version = match observed { + Ok(version) => *version, + Err(why) => panic!( + "a probe sent in round {round}, while the next hop was moving, is not \ + attributable to any version: {why}" + ), + }; + if last { + assert_eq!( + version, churn, + "a probe sent after the churn had finished was forwarded by version \ + {version}, not by version {churn}, the last one published. Every \ + publish returned before the barrier that released this probe, so a \ + reader still serving an earlier version is serving a next hop that \ + no longer exists" + ); + continue; + } + assert!( + version == round || version + 1 == round, + "a probe sent in round {round} was forwarded by version {version}. \ + Version {} was published before this round opened, so no reader may \ + still be serving anything older, and version {round} is the newest that \ + exists", + round - 1 ); - continue; - } - assert!( - version == round || version + 1 == round, - "a probe sent in round {round} was forwarded by version {version}. \ - Version {} was published before this round opened, so no reader may \ - still be serving anything older, and version {round} is the newest that \ - exists", - round - 1 - ); - if version == round { - FRESH.fetch_add(1, Ordering::Relaxed); - } else { - STALE.fetch_add(1, Ordering::Relaxed); + if version == round { + FRESH.fetch_add(1, Ordering::Relaxed); + } else { + STALE.fetch_add(1, Ordering::Relaxed); + } } } - } - }); + }); + }); let (fresh, stale) = (FRESH.load(Ordering::Relaxed), STALE.load(Ordering::Relaxed)); eprintln!("fresh={fresh} stale={stale}"); @@ -5427,7 +5583,8 @@ mod model { #[concurrency::model_test] fn re_enacting_a_configuration_under_load_disturbs_nothing() { - const FLOWS: u8 = 2; + const CASES: usize = 64; + const APPLIES: u8 = 4; static COMPLETED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); @@ -5446,38 +5603,172 @@ mod model { }; let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); - concurrency::stress(move || { - let overlay = overlay_with_exposes_and_acl(exposes(), None) - .expect("the fixture exposes form an overlay") - .validate() - .expect("the fixture overlay validates"); - let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); - let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); - let blueprint = fleet.blueprint(); - let entering = handle.clone(); - let gate = Arc::new(concurrency::sync::Barrier::new(3)); - let mut reports = Vec::new(); + bolero::check!() + .with_max_len(MAX_INPUT_LEN) + .with_type() + .cloned() + .with_iterations(CASES) + .for_each(|(flows, host): (u8, u8)| { + let flows = flows % 4 + 1; + let src_host = host % 8; + let handle = handle.clone(); + concurrency::stress(move || { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let entering = handle.clone(); + let gate = Arc::new(concurrency::sync::Barrier::new(3)); + let mut reports = Vec::new(); + + thread::scope(|scope| { + let running: Vec<_> = (0..2u8) + .map(|which| { + let entering = entering.clone(); + let gate = gate.clone(); + thread::Builder::new() + .name(format!("tenant-{which}")) + .spawn_scoped(scope, move || { + let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); + let _evidence = + tracectl::evidence::capture(format!("tenant-{which}")); + let mut worker = blueprint.worker(); + gate.wait(); + let mut seen = Vec::new(); + for round in 1..=APPLIES { + for nth in 0..flows { + seen.push(( + round, + without_unwinding(|| { + let src = format!("1.1.{which}.{}", src_host + nth + 1); + let mut convo = Conversation::new( + super::routed::Path::fixture(), + src.parse().unwrap_or_else(|e| { + unreachable!("{src}: {e}") + }), + "3.3.3.1" + .parse() + .unwrap_or_else(|e| unreachable!("{e}")), + u16::from(round) * 100 + u16::from(nth) + 1000, + 80, + ); + drive(&mut worker, &mut convo); + (convo.checked(), convo.describe()) + }), + )); + } + gate.wait(); + } + seen + }) + .expect("spawn tenant") + }) + .collect(); - thread::scope(|scope| { - let running: Vec<_> = (0..2u8) - .map(|which| { + gate.wait(); + for _ in 0..APPLIES { + fleet.reconfigure(&overlay); + ENACTED.fetch_add(1, Ordering::Relaxed); + gate.wait(); + } + + for worker in running { + reports.push(worker.join().expect("tenant panicked")); + } + }); + + for seen in reports { + for (round, ran) in seen { + let (checked, described) = ran.unwrap_or_else(|why| { + panic!("a conversation in round {round} panicked mid-enactment: {why}") + }); + assert!( + checked, + "a conversation in round {round} did not survive the configuration it was \ + already running being enacted again. Every enactment before this round \ + had returned, and the one racing it changes nothing. {described}" + ); + COMPLETED.fetch_add(1, Ordering::Relaxed); + } + } + }); + }); + + let (completed, enacted) = ( + COMPLETED.load(Ordering::Relaxed), + ENACTED.load(Ordering::Relaxed), + ); + eprintln!("completed={completed} enacted={enacted}"); + super::assert_covered( + completed > 0 && enacted > 0, + "either no conversation completed or no configuration was enacted, so nothing was raced", + ); + } + + #[concurrency::model_test] + fn the_cli_can_be_read_while_the_dataplane_works() { + const CASES: usize = 64; + + const APPLIES: u8 = 3; + + static COMPLETED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + static ANSWERED: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + let _eal = dpdk::test_support::start_eal(); + + let rt = cfg_select! { + feature = "shuttle" => None::, + _ => Some( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + ) + }; + let handle = rt.as_ref().map(tokio::runtime::Runtime::handle).cloned(); + + bolero::check!() + .with_max_len(MAX_INPUT_LEN) + .with_type() + .cloned() + .with_iterations(CASES) + .for_each(|(flows, host): (u8, u8)| { + let flows = flows % 4 + 1; + let src_host = host % 8; + let handle = handle.clone(); + concurrency::stress(move || { + let overlay = overlay_with_exposes_and_acl(exposes(), None) + .expect("the fixture exposes form an overlay") + .validate() + .expect("the fixture overlay validates"); + let tables = topology(&[vni(LOCAL_VNI), vni(REMOTE_VNI)]); + let fleet = Fleet::lowering(&overlay, Some(&tables), Arc::new(FlowTable::default())); + let blueprint = fleet.blueprint(); + let readers = blueprint.cli_readers(); + let entering = handle.clone(); + let gate = Arc::new(concurrency::sync::Barrier::new(3)); + + let (seen, read) = thread::scope(|scope| { + let forwarding = { let entering = entering.clone(); let gate = gate.clone(); thread::Builder::new() - .name(format!("tenant-{which}")) + .name("tenant".to_string()) .spawn_scoped(scope, move || { let _guard = entering.as_ref().map(tokio::runtime::Handle::enter); - let _evidence = - tracectl::evidence::capture(format!("tenant-{which}")); + let _evidence = tracectl::evidence::capture("tenant"); let mut worker = blueprint.worker(); gate.wait(); let mut seen = Vec::new(); for round in 1..=APPLIES { - for nth in 0..FLOWS { + for nth in 0..flows { seen.push(( round, without_unwinding(|| { - let src = format!("1.1.{which}.{}", nth + 1); + let src = format!("1.1.0.{}", src_host + nth + 1); let mut convo = Conversation::new( super::routed::Path::fixture(), src.parse().unwrap_or_else(|e| { @@ -5499,45 +5790,79 @@ mod model { seen }) .expect("spawn tenant") - }) - .collect(); + }; + + let reading = { + let gate = gate.clone(); + thread::Builder::new() + .name("cli".to_string()) + .spawn_scoped(scope, move || { + let _evidence = tracectl::evidence::capture("cli"); + gate.wait(); + let mut answered = Vec::new(); + for round in 1..=APPLIES { + for nth in 0..(2 * readers.len()) { + answered.push(( + round, + without_unwinding(|| { + let (command, text) = readers.read_one(nth); + (command, text.len()) + }), + )); + } + gate.wait(); + } + answered + }) + .expect("spawn cli") + }; - gate.wait(); - for _ in 0..APPLIES { - fleet.reconfigure(&overlay); - ENACTED.fetch_add(1, Ordering::Relaxed); gate.wait(); - } + for _ in 0..APPLIES { + fleet.reconfigure(&overlay); + gate.wait(); + } + + ( + forwarding.join().expect("tenant panicked"), + reading.join().expect("cli panicked"), + ) + }); - for worker in running { - reports.push(worker.join().expect("tenant panicked")); + for (round, answer) in read { + let (command, length) = answer.unwrap_or_else(|why| { + panic!("`{why}` while answering a cli command in round {round}") + }); + assert!( + length > 0, + "`{command}` answered with nothing in round {round}, so the reader reached \ + the state but could not say anything about it" + ); + ANSWERED.fetch_add(1, Ordering::Relaxed); } - }); - for seen in reports { for (round, ran) in seen { let (checked, described) = ran.unwrap_or_else(|why| { - panic!("a conversation in round {round} panicked mid-enactment: {why}") + panic!("a conversation in round {round} panicked while the cli read: {why}") }); assert!( checked, - "a conversation in round {round} did not survive the configuration it was \ - already running being enacted again. Every enactment before this round \ - had returned, and the one racing it changes nothing. {described}" + "a conversation in round {round} did not survive the cli being read beside \ + it. Reading is supposed to be an observation, not a change. {described}" ); COMPLETED.fetch_add(1, Ordering::Relaxed); } - } - }); + }); + }); - let (completed, enacted) = ( + let (completed, answered) = ( COMPLETED.load(Ordering::Relaxed), - ENACTED.load(Ordering::Relaxed), + ANSWERED.load(Ordering::Relaxed), ); - eprintln!("completed={completed} enacted={enacted}"); + eprintln!("completed={completed} answered={answered}"); super::assert_covered( - completed > 0 && enacted > 0, - "either no conversation completed or no configuration was enacted, so nothing was raced", + completed > 0 && answered > 0, + "either no conversation completed or no cli command was answered, so nothing was raced", ); } diff --git a/flow-entry/src/flow_table/concurrent_fuzz.rs b/flow-entry/src/flow_table/concurrent_fuzz.rs index 78924a974e..7fd0aaef71 100644 --- a/flow-entry/src/flow_table/concurrent_fuzz.rs +++ b/flow-entry/src/flow_table/concurrent_fuzz.rs @@ -337,23 +337,20 @@ impl Scenario { /// `just test sanitize=thread`), or the full portfolio under shuttle. #[concurrency::model_test] fn stress_test_concurrency_model() { - // Single-threaded runtime is enough: we never need the timer task to - // run, only a context for `insert`'s `tokio::task::spawn` to succeed. - let rt = cfg_select! { - feature = "shuttle" => None::, - _ => Some( - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("build tokio runtime") - ) - }; - let handle = rt.as_ref().map(|rt| rt.handle().clone()); bolero::check!() .with_type() .cloned() .for_each(|scenario: Scenario| { - let handle = handle.clone(); + let rt = cfg_select! { + feature = "shuttle" => None::, + _ => Some( + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build tokio runtime") + ) + }; + let handle = rt.as_ref().map(|rt| rt.handle().clone()); concurrency::stress(move || { scenario.run(handle.as_ref()); }); diff --git a/flow-entry/src/flow_table/table.rs b/flow-entry/src/flow_table/table.rs index a1cb273477..70800a3723 100644 --- a/flow-entry/src/flow_table/table.rs +++ b/flow-entry/src/flow_table/table.rs @@ -901,8 +901,7 @@ mod tests { Some(VpcDiscriminant::VNI( Vni::new_checked(u32::from(i) + 1).unwrap(), )), - format!("10.0.{i}.1").parse::().unwrap(), - format!("10.0.{i}.2").parse::().unwrap(), + v4_addrs(&format!("10.0.{i}.1"), &format!("10.0.{i}.2")), IpProtoKey::Tcp(TcpProtoKey { src_port: TcpPort::new_checked(1000 + i).unwrap(), dst_port: TcpPort::new_checked(2000 + i).unwrap(), diff --git a/flow-filter/src/context/tables.rs b/flow-filter/src/context/tables.rs index 1c4a72fc64..9bda417f39 100644 --- a/flow-filter/src/context/tables.rs +++ b/flow-filter/src/context/tables.rs @@ -38,8 +38,6 @@ use acl::dpdk::lookup::{DpdkAclLookup, MAX_BATCH}; use acl::dpdk::rule::{AclFieldChunks, RuleSpec}; #[cfg(test)] use acl::reference::table::{RefRule, ReferenceTable}; -use concurrency::sync::LazyLock; -use concurrency::sync::atomic::{AtomicU64, Ordering}; use config::external::overlay::ValidatedOverlay; use dpdk::acl::{CategoryMask, Priority}; #[cfg(test)] @@ -348,6 +346,9 @@ impl fmt::Debug for AnyTable { } concurrency::with_std! { + use concurrency::sync::LazyLock; + use concurrency::sync::atomic::{AtomicU64, Ordering}; + static TABLE_SEQ: LazyLock = LazyLock::new(|| AtomicU64::new(0)); fn next_in_sequence() -> u64 { diff --git a/interface-manager/src/interface/mod.rs b/interface-manager/src/interface/mod.rs index 6dfcac96b9..95a18d7509 100644 --- a/interface-manager/src/interface/mod.rs +++ b/interface-manager/src/interface/mod.rs @@ -894,7 +894,19 @@ mod tests { bolero::check!().with_type().for_each( |(requirement, observation): &(InterfaceSpec, Interface)| { if requirement == observation { - assert_eq!(requirement, &observation.as_requirement().unwrap()); + let observed = observation + .as_requirement() + .unwrap_or_else(|| unreachable!("it matched, so it has a requirement")); + assert_eq!(requirement.name, observed.name); + assert_eq!(requirement.admin_state, observed.admin_state); + assert_eq!(requirement.controller, observed.controller); + assert_eq!(requirement.properties, observed.properties); + if requirement.mac.is_some() { + assert_eq!(requirement.mac, observed.mac); + } + if requirement.mtu.is_some() { + assert_eq!(requirement.mtu, observed.mtu); + } } else { match observation.as_requirement() { None => {} diff --git a/mgmt/src/processor/proc.rs b/mgmt/src/processor/proc.rs index 84b76ab33e..c245144e3d 100644 --- a/mgmt/src/processor/proc.rs +++ b/mgmt/src/processor/proc.rs @@ -238,9 +238,11 @@ impl ConfigProcessor { let stats_store = &self.proc_params.vpc_stats_store; - let names = stats_store.snapshot_names().await; - let pair_snap = stats_store.snapshot_pairs().await; - let vpc_snap = stats_store.snapshot_vpcs().await; + let stats::StatsSnapshot { + names, + pairs: pair_snap, + vpcs: vpc_snap, + } = stats_store.snapshot().await; // Helper to check if a flow stats has any traffic #[inline] diff --git a/nat/src/masquerade/fuzz.rs b/nat/src/masquerade/fuzz.rs index 51f5ce4fc0..2cae2ece27 100644 --- a/nat/src/masquerade/fuzz.rs +++ b/nat/src/masquerade/fuzz.rs @@ -43,12 +43,25 @@ impl ValueGenerator for Scenario { fn settled(body: impl FnOnce()) { const PAST_ANY_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(30); + const GIVE_UP: usize = 4096; // nosemgrep: rust-no-direct-std-sync-import static CLOCK: std::sync::LazyLock = std::sync::LazyLock::new(clock::virtual_time::Paused::new); // nosemgrep: rust-no-direct-std-sync-import CLOCK.block_on(async { body(); clock::virtual_time::advance(PAST_ANY_TIMEOUT).await; + let handle = tokio::runtime::Handle::current(); + for _ in 0..GIVE_UP { + if handle.metrics().num_alive_tasks() == 0 { + return; + } + tokio::task::yield_now().await; + } + panic!( + "{} tasks from this case would not retire; the flow tables they hold will accumulate \ + until the run is out of memory", + handle.metrics().num_alive_tasks() + ); }); } diff --git a/nat/src/portfw/fuzz.rs b/nat/src/portfw/fuzz.rs index d6d5768332..643ef96abd 100644 --- a/nat/src/portfw/fuzz.rs +++ b/nat/src/portfw/fuzz.rs @@ -42,12 +42,25 @@ impl ValueGenerator for Scenario { fn settled(body: impl FnOnce()) { const PAST_ANY_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(30); + const GIVE_UP: usize = 4096; // nosemgrep: rust-no-direct-std-sync-import static CLOCK: std::sync::LazyLock = std::sync::LazyLock::new(clock::virtual_time::Paused::new); // nosemgrep: rust-no-direct-std-sync-import CLOCK.block_on(async { body(); clock::virtual_time::advance(PAST_ANY_TIMEOUT).await; + let handle = tokio::runtime::Handle::current(); + for _ in 0..GIVE_UP { + if handle.metrics().num_alive_tasks() == 0 { + return; + } + tokio::task::yield_now().await; + } + panic!( + "{} tasks from this case would not retire; the flow tables they hold will accumulate \ + until the run is out of memory", + handle.metrics().num_alive_tasks() + ); }); } diff --git a/net/src/flows/flow_info_fuzz.rs b/net/src/flows/flow_info_fuzz.rs index 9c68a5e861..6f50951437 100644 --- a/net/src/flows/flow_info_fuzz.rs +++ b/net/src/flows/flow_info_fuzz.rs @@ -407,15 +407,16 @@ fn a_genid_is_remembered_and_reaches_the_partner() { return; }; + let untouched = second.genid(); first.set_genid(*genid); assert_eq!( first.genid(), *genid, "a genid must read back as it was set" ); - assert_ne!( + assert_eq!( second.genid(), - *genid, + untouched, "setting one half's genid must not reach the other" ); diff --git a/stats/Cargo.toml b/stats/Cargo.toml index ffbada8f22..e9d2aacb1a 100644 --- a/stats/Cargo.toml +++ b/stats/Cargo.toml @@ -36,7 +36,7 @@ tracing = { workspace = true, features = ["attributes"] } [dev-dependencies] clock = { workspace = true, features = ["virtual"] } -tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] } +tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread", "sync", "time"] } bolero = { workspace = true } net = { workspace = true, features = ["bolero"] } vpcmap = { workspace = true, features = ["bolero"] } diff --git a/stats/src/dpstats.rs b/stats/src/dpstats.rs index 10f9264183..79d769818a 100644 --- a/stats/src/dpstats.rs +++ b/stats/src/dpstats.rs @@ -10,7 +10,7 @@ use pipeline::NetworkFunction; use concurrency::sync::Arc; use kanal::ReceiveError; -use std::collections::{HashMap, HashSet, VecDeque}; +use std::collections::{BTreeSet, HashMap, HashSet, VecDeque}; use std::time::{Duration, Instant}; use vpcmap::VpcDiscriminant; use vpcmap::map::VpcMapReader; @@ -77,7 +77,19 @@ fn apportion_into_batches( value: PacketAndByte, mut apply: impl FnMut(&mut TransmitSummary, PacketAndByte), ) { - if (value.packets == 0 && value.bytes == 0) || total_ov == 0 { + if value.packets == 0 && value.bytes == 0 { + return; + } + if total_ov == 0 { + let Some(batch) = slices.first_mut() else { + error!( + "no open batch for an update covering {} packets: the schedule has fallen behind", + value.packets + ); + return; + }; + let tx = batch.vpc.entry(src).or_insert_with(TransmitSummary::new); + apply(tx, value); return; } let mut rem_pkts = value.packets; @@ -117,16 +129,18 @@ fn apportion_into_batches( } /// Take a synchronous snapshot of `(disc, name)` pairs from the VPC map reader. -fn snapshot_vpc_pairs(reader: &VpcMapReader) -> Vec<(VpcDiscriminant, String)> { +fn snapshot_vpc_pairs(reader: &VpcMapReader) -> Option> { match reader.enter() { - Some(guard) => guard - .0 - .values() - .map(|VpcMapName { disc, name }| (*disc, name.clone())) - .collect(), + Some(guard) => Some( + guard + .0 + .values() + .map(|VpcMapName { disc, name }| (*disc, name.clone())) + .collect(), + ), None => { - warn!("vpcmap reader guard acquisition failed; proceeding with empty snapshot"); - Vec::new() + warn!("vpcmap reader guard acquisition failed; leaving the collector's view unchanged"); + None } } } @@ -152,10 +166,34 @@ fn set_gauges_to_zero(base: &str, labels: Vec<(String, String)>) { } } -/// Zero the `vpc_*` gauge family (per-VPC totals/drops and per-pair traffic) for the given labels. -#[inline] -fn set_vpc_gauges_to_zero(labels: Vec<(String, String)>) { - set_gauges_to_zero("vpc", labels); +fn register_series( + names: &[(VpcDiscriminant, String)], +) -> hashbrown::HashMap { + let vpc_data = names + .iter() + .map(|(disc, name)| (*disc, name.clone(), vec![])) + .collect::>(); + VpcMetricsSpec::new(vpc_data) + .into_iter() + .map(|(disc, spec)| (disc, spec.build())) + .collect() +} + +fn exported_series(names: &BTreeSet) -> BTreeSet<(&'static str, Vec<(String, String)>)> { + let mut series = BTreeSet::new(); + for name in names { + series.insert(("vpc", vec![("total".to_string(), name.clone())])); + series.insert(("vpc", vec![("drops".to_string(), name.clone())])); + for other in names { + let pair = vec![ + ("from".to_string(), name.clone()), + ("to".to_string(), other.clone()), + ]; + series.insert(("vpc", pair.clone())); + series.insert((PAIR_DROPS_METRIC_BASE, pair)); + } + } + series } /// A `StatsCollector` is responsible for collecting and aggregating packet statistics for a @@ -178,14 +216,13 @@ pub struct StatsCollector { /// Shared store for snapshots/rates usable by gRPC, CLI, etc. vpc_store: Arc, alive_vpcs: HashSet, - /// `known` is a reference to the previous snapshot of `alive` VPCs, used to detect removals. - known_vpcs: HashSet, known_names: HashMap, } impl StatsCollector { const DEFAULT_CHANNEL_CAPACITY: usize = 256; const TIME_TICK: Duration = Duration::from_secs(1); + const OUTSTANDING: usize = 10; #[tracing::instrument(level = "info")] pub fn new(vpcmap_r: VpcMapReader) -> (StatsCollector, PacketStatsWriter) { @@ -203,50 +240,26 @@ impl StatsCollector { ) -> (StatsCollector, PacketStatsWriter, Arc) { let (s, r) = kanal::bounded(Self::DEFAULT_CHANNEL_CAPACITY); - // Snapshot current VPC names from the reader to seed metric registrations - let vpc_data = match vpcmap_r.enter() { - Some(guard) => guard - .0 - .values() - .map(|VpcMapName { disc, name }| { - ( - *disc, - name.clone(), - vec![("from".to_string(), name.clone())], - ) - }) - .collect::>(), - None => { - warn!( - "vpcmap reader guard acquisition failed during initialization; seeding empty metrics" - ); - Vec::new() - } - }; - - let name_pairs = snapshot_vpc_pairs(&vpcmap_r); + let name_pairs = snapshot_vpc_pairs(&vpcmap_r).unwrap_or_default(); vpc_store.set_many_vpc_names_sync(name_pairs.clone()); let alive_vpcs: HashSet = - vpc_data.iter().map(|(disc, _, _)| *disc).collect(); + name_pairs.iter().map(|(disc, _)| *disc).collect(); let mut known_names: HashMap = HashMap::new(); - for (disc, name) in name_pairs { + for (disc, name) in name_pairs.iter().cloned() { known_names.insert(disc, name); } - let known_vpcs = alive_vpcs.clone(); - let metrics = VpcMetricsSpec::new(vpc_data) - .into_iter() - .map(|(disc, spec)| (disc, spec.build())) - .collect(); + let metrics = register_series(&name_pairs); let updates = PacketStatsReader(r); - let outstanding: VecDeque<_> = (0..10) - .scan( - BatchSummary::::new(clock::now() + Self::TIME_TICK), - |prior, _| Some(BatchSummary::new(prior.planned_end + Self::TIME_TICK)), - ) + let outstanding: VecDeque<_> = (0..Self::OUTSTANDING) + .scan(clock::now(), |start, _| { + let batch = BatchSummary::::with_start(*start, Self::TIME_TICK); + *start += Self::TIME_TICK; + Some(batch) + }) .collect(); let store_clone = Arc::clone(&vpc_store); @@ -259,87 +272,57 @@ impl StatsCollector { updates, vpc_store, alive_vpcs, - known_vpcs, known_names, }; let writer = PacketStatsWriter(s); (stats, writer, store_clone) } - /// Update the list of VPCs known to the stats collector (sync snapshot; no awaits). - #[tracing::instrument(level = "debug")] - fn refresh(&mut self) -> impl Iterator { - let pairs = snapshot_vpc_pairs(&self.vpcmap_r); // Vec<(disc, name)> - // persist names for gRPC/others (no await) - self.vpc_store.set_many_vpc_names_sync(pairs.clone()); - - let vpc_data = pairs - .into_iter() - .map(|(disc, name)| (disc, name, vec![])) - .collect::>(); - - VpcMetricsSpec::new(vpc_data) - .into_iter() - .map(|(disc, spec)| (disc, spec.build())) - } - #[tracing::instrument(level = "debug")] async fn refresh_vpc_store(&mut self) { - let pairs = snapshot_vpc_pairs(&self.vpcmap_r); - self.vpc_store.set_many_vpc_names_sync(pairs.clone()); + let Some(pairs) = snapshot_vpc_pairs(&self.vpcmap_r) else { + return; + }; + let new_names: HashMap = pairs.iter().cloned().collect(); - let new_alive: HashSet = pairs.iter().map(|(d, _)| *d).collect(); + if new_names == self.known_names { + return; + } - let mut removed: Vec = - self.known_vpcs.difference(&new_alive).copied().collect(); - removed.sort(); + self.alive_vpcs = pairs.iter().map(|(disc, _)| *disc).collect(); - for (disc, name) in &pairs { - self.known_names.insert(*disc, name.clone()); + let handovers: Vec<(VpcDiscriminant, String)> = new_names + .iter() + .filter(|(disc, name)| { + self.known_names + .get(disc) + .is_some_and(|previously| previously != *name) + }) + .map(|(disc, name)| (*disc, name.clone())) + .collect(); + if !handovers.is_empty() { + let recycled: HashSet = + handovers.iter().map(|(disc, _)| *disc).collect(); + self.vpc_store.hand_over(&handovers).await; + self.outstanding + .iter_mut() + .for_each(|batch| batch.forget(&recycled)); + self.submitted + .each_sample_mut(|sample| forget_from(sample, &recycled)); } - self.alive_vpcs = new_alive.clone(); - - // prune any removed VPCs / pairs so they do not show up in snapshots/status self.vpc_store.prune_to_vpcs(&self.alive_vpcs).await; + self.vpc_store.set_many_vpc_names_sync(pairs.clone()); - if !removed.is_empty() { - let mut alive_names: Vec = pairs.iter().map(|(_, n)| n.clone()).collect(); - alive_names.sort(); - alive_names.dedup(); - - for disc in removed { - let removed_name = self - .known_names - .get(&disc) - .cloned() - .unwrap_or_else(|| format!("{disc:?}")); - - // total/drops series for the removed VPC - set_vpc_gauges_to_zero(vec![("total".to_string(), removed_name.clone())]); - set_vpc_gauges_to_zero(vec![("drops".to_string(), removed_name.clone())]); - - // peering series (traffic and per-pair drops) in both directions - for other_name in &alive_names { - let fwd = vec![ - ("from".to_string(), removed_name.clone()), - ("to".to_string(), other_name.clone()), - ]; - let rev = vec![ - ("from".to_string(), other_name.clone()), - ("to".to_string(), removed_name.clone()), - ]; - set_vpc_gauges_to_zero(fwd.clone()); - set_vpc_gauges_to_zero(rev.clone()); - set_gauges_to_zero(PAIR_DROPS_METRIC_BASE, fwd); - set_gauges_to_zero(PAIR_DROPS_METRIC_BASE, rev); - } - - self.known_names.remove(&disc); - } + let was: BTreeSet = self.known_names.values().cloned().collect(); + let now: BTreeSet = new_names.values().cloned().collect(); + for (base, labels) in exported_series(&was).difference(&exported_series(&now)) { + set_gauges_to_zero(base, labels.clone()); } - self.known_vpcs = new_alive; + self.metrics = register_series(&pairs); + + self.known_names = new_names; } /// Run the collector (async). Does not return if awaited. @@ -382,9 +365,6 @@ impl StatsCollector { async fn update(&mut self, update: Option) { self.refresh_vpc_store().await; if let Some(update) = update { - // Refresh Prometheus registrations based on the current VPC snapshot. - self.metrics = self.refresh().collect(); - // Find outstanding changes which line up with batch let mut slices: Vec<_> = self .outstanding @@ -398,92 +378,6 @@ impl StatsCollector { }) .collect(); - // Proportionally distribute each (src,dst) update across overlapping batches. - update.summary.vpc.iter().for_each(|(src, summary)| { - summary.dst.iter().for_each(|(dst, stats)| { - if stats.packets == 0 && stats.bytes == 0 { - return; - } - - let upd_start = update.summary.start; - let upd_end = update.start() + update.duration; - - // Pre-compute overlaps with all candidate batch slices - let overlaps: Vec = slices - .iter() - .map(|b| overlap_nanos(b.start, b.planned_end, upd_start, upd_end)) - .collect(); - let total_ov: u128 = overlaps.iter().copied().sum(); - if total_ov == 0 { - return; - } - - // Integer-safe split: give the remainder to the last overlapping bucket - let mut rem_pkts = stats.packets; - let mut rem_bytes = stats.bytes; - - let last_idx = overlaps - .iter() - .enumerate() - .rfind(|&(_, &ov)| ov > 0) - .map(|(i, _)| i); - - for (i, batch) in slices.iter_mut().enumerate() { - let ov = overlaps[i]; - if ov == 0 { - continue; - } - - let is_last = Some(i) == last_idx; - - let pkts_in = if is_last { - rem_pkts - } else { - let v = ((stats.packets as u128) * ov / total_ov) as u64; - rem_pkts = rem_pkts.saturating_sub(v); - v - }; - - let bytes_in = if is_last { - rem_bytes - } else { - let v = ((stats.bytes as u128) * ov / total_ov) as u64; - rem_bytes = rem_bytes.saturating_sub(v); - v - }; - - if pkts_in == 0 && bytes_in == 0 { - continue; - } - - let apportioned = PacketAndByte { - packets: pkts_in, - bytes: bytes_in, - }; - - match batch.vpc.get_mut(src) { - None => { - let mut tx_summary = TransmitSummary::new(); - tx_summary.dst.insert(*dst, apportioned); - batch.vpc.insert(*src, tx_summary); - } - Some(tx_summary) => match tx_summary.dst.get_mut(dst) { - None => { - tx_summary.dst.insert(*dst, apportioned); - } - Some(s) => { - *s += apportioned; - } - }, - } - } - }); - }); - - // Drops are collected per source (a total that also includes drops whose destination - // VPC could not be resolved) and per (src,dst) pair. Neither is rate-smoothed, but both - // must reach `submit_expired` via the outstanding batches, so apportion them across the - // same overlapping slices as forward traffic. let upd_start = update.summary.start; let upd_end = update.start() + update.duration; let overlaps: Vec = slices @@ -497,6 +391,26 @@ impl StatsCollector { .rfind(|&(_, &ov)| ov > 0) .map(|(i, _)| i); + // Proportionally distribute each (src,dst) update across overlapping batches. + update.summary.vpc.iter().for_each(|(src, summary)| { + summary.dst.iter().for_each(|(dst, stats)| { + let dst = *dst; + apportion_into_batches( + &mut slices, + &overlaps, + total_ov, + last_idx, + *src, + *stats, + |tx, v| add_into_map(&mut tx.dst, dst, v), + ); + }); + }); + + // Drops are collected per source (a total that also includes drops whose destination + // VPC could not be resolved) and per (src,dst) pair. Neither is rate-smoothed, but both + // must reach `submit_expired` via the outstanding batches, so apportion them across the + // same overlapping slices as forward traffic. update.summary.vpc.iter().for_each(|(src, summary)| { for (dst, drops) in summary.pair_drops.iter() { let dst = *dst; @@ -570,16 +484,16 @@ impl StatsCollector { let mut total_bytes = 0u64; for (&dst, &stats) in tx_summary.dst.iter() { + total_pkts = total_pkts.saturating_add(stats.packets); + total_bytes = total_bytes.saturating_add(stats.bytes); + if !self.alive_vpcs.contains(&dst) { - debug!("skipping stats for removed VPC {dst}"); + debug!("skipping pair stats for removed VPC {dst}"); continue; } self.vpc_store .add_pair_counts(src, dst, stats.packets, stats.bytes) .await; - - total_pkts = total_pkts.saturating_add(stats.packets); - total_bytes = total_bytes.saturating_add(stats.bytes); } if total_pkts != 0 || total_bytes != 0 { @@ -613,6 +527,11 @@ impl StatsCollector { // Refresh count gauges from the store (so reuse doesn't carry stale totals). let pair_snap = self.vpc_store.snapshot_pairs().await; + let mut carried: hashbrown::HashMap> = + hashbrown::HashMap::new(); + for &((src, dst), _) in &pair_snap { + carried.entry(src).or_default().insert(dst); + } for ((src, dst), fs) in pair_snap { if !self.alive_vpcs.contains(&src) || !self.alive_vpcs.contains(&dst) { continue; @@ -688,23 +607,23 @@ impl StatsCollector { // Smoothed entry for this src (if any) let maybe_tx = smoothed_by_src.get(&src); - // For every known dst under this src, either set smoothed rate or zero. - for (&dst, action) in metrics.peering.iter() { + let mut publish: BTreeSet = + carried.get(&src).cloned().unwrap_or_default(); + if let Some(tx_summary) = maybe_tx { + publish.extend(tx_summary.dst.iter().map(|(dst, _)| *dst)); + } + + for dst in publish { if !self.alive_vpcs.contains(&dst) { debug!("skipping rate stats for removed VPC {dst}"); continue; } - let (pps, bps) = if let Some(tx_summary) = maybe_tx { - if let Some(rate) = tx_summary.dst.get(&dst) { - (rate.packets, rate.bytes) - } else { - // zero if pair absent in window - (0.0, 0.0) - } - } else { - // here as well - (0.0, 0.0) + let Some(action) = metrics.peering.get(&dst) else { + continue; }; + let (pps, bps) = maybe_tx + .and_then(|tx_summary| tx_summary.dst.get(&dst)) + .map_or((0.0, 0.0), |rate| (rate.packets, rate.bytes)); // Export to Prometheus gauges action.tx.packet.rate.metric.set(pps); @@ -787,6 +706,28 @@ pub struct TransmitSummary { } const SMALL_MAP_CAPACITY: usize = 8; + +impl TransmitSummary { + fn forget(&mut self, discs: &HashSet) { + self.dst.retain(|dst, _| !discs.contains(dst)); + self.pair_drops.retain(|dst, _| !discs.contains(dst)); + } +} + +fn forget_from( + vpc: &mut hashbrown::HashMap>, + discs: &HashSet, +) { + vpc.retain(|src, _| !discs.contains(src)); + vpc.values_mut().for_each(|tx| tx.forget(discs)); +} + +impl BatchSummary { + fn forget(&mut self, discs: &HashSet) { + forget_from(&mut self.vpc, discs); + } +} + impl TransmitSummary { pub fn new() -> Self where @@ -929,10 +870,15 @@ impl NetworkFunction for Stats { )); let duration = time.duration_since(self.update.start); let summary = std::mem::replace(&mut self.update, batch); - let update = MetricsUpdate { duration, summary }; - match self.stats.0.try_send(update) { + let mut update = Some(MetricsUpdate { duration, summary }); + match self.stats.0.try_send_option(&mut update) { Ok(true) => trace!("sent stats update"), - Ok(false) => warn!("metrics channel full! Some metrics lost"), + Ok(false) => { + let held = update.unwrap_or_else(|| unreachable!()).summary; + self.update = held; + self.update.planned_end = time + self.delivery_schedule; + warn!("metrics channel full; holding this batch open until it can be sent"); + } Err(err) => { error!("{err}"); panic!("{err}"); @@ -1214,6 +1160,95 @@ mod drop_stats_tests { let _drained: Vec<_> = stats.process(packets.into_iter()).collect(); } + #[test] + fn a_batch_closes_on_schedule_with_no_packets_to_process() { + let (a, b) = (vpcd(100), vpcd(200)); + let (sender, receiver) = kanal::bounded(4); + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let tick = Duration::from_secs(1); + let mut stats = Stats::with_delivery_schedule("test", PacketStatsWriter(sender), tick); + run( + &mut stats, + vec![mk_packet(Some(a), Some(b), Some(DoneReason::Delivered))], + ); + assert!( + matches!(receiver.try_recv(), Ok(None)), + "the batch was sent before its schedule was up" + ); + + clock::virtual_time::advance(tick * 2).await; + run(&mut stats, vec![]); + + let Ok(Some(update)) = receiver.try_recv() else { + panic!("the batch did not close, so the packet in it is not counted anywhere"); + }; + assert_eq!( + update + .summary + .vpc + .get(&a) + .and_then(|tx| tx.dst.get(&b)) + .map(|counts| counts.packets), + Some(1) + ); + }); + } + + #[test] + fn a_batch_that_cannot_be_sent_is_held_rather_than_dropped() { + const FED: usize = 5; + let (a, b) = (vpcd(100), vpcd(200)); + let (sender, receiver) = kanal::bounded(1); + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let tick = Duration::from_secs(1); + let mut stats = Stats::with_delivery_schedule("test", PacketStatsWriter(sender), tick); + + for _ in 0..FED { + clock::virtual_time::advance(tick * 2).await; + run( + &mut stats, + vec![mk_packet(Some(a), Some(b), Some(DoneReason::Delivered))], + ); + } + + let mut sent = 0u64; + for _ in 0..FED * 2 { + while let Ok(Some(update)) = receiver.try_recv() { + sent += update + .summary + .vpc + .get(&a) + .and_then(|tx| tx.dst.get(&b)) + .map_or(0, |counts| counts.packets); + } + clock::virtual_time::advance(tick * 2).await; + run(&mut stats, vec![]); + } + while let Ok(Some(update)) = receiver.try_recv() { + sent += update + .summary + .vpc + .get(&a) + .and_then(|tx| tx.dst.get(&b)) + .map_or(0, |counts| counts.packets); + } + + let still_held = stats + .update + .vpc + .get(&a) + .and_then(|tx| tx.dst.get(&b)) + .map_or(0, |counts| counts.packets); + assert_eq!( + sent + still_held, + FED as u64, + "{FED} packets counted, {sent} sent and {still_held} still in hand" + ); + }); + } + #[test] fn delivered_pair_counts_forward_only() { let (a, b) = (vpcd(100), vpcd(200)); @@ -1410,7 +1445,7 @@ mod drop_stats_tests { } #[test] - fn apportion_no_overlap_records_nothing() { + fn apportion_no_overlap_still_counts() { let mut b0 = batch(1); let mut slices: Vec<&mut BatchSummary> = vec![&mut b0]; let src = vpcd(9); @@ -1428,6 +1463,918 @@ mod drop_stats_tests { |tx, v| tx.drops += v, ); drop(slices); - assert!(b0.vpc.is_empty()); + let recorded = b0.vpc.get(&src).map(|tx| tx.drops); + assert_eq!( + recorded, + Some(PacketAndByte { + packets: 5, + bytes: 50 + }) + ); + } + + #[test] + fn apportion_with_no_open_batch_records_nothing() { + let mut slices: Vec<&mut BatchSummary> = vec![]; + apportion_into_batches( + &mut slices, + &[], + 0, + None, + vpcd(9), + PacketAndByte { + packets: 5, + bytes: 50, + }, + |tx, v| tx.drops += v, + ); + } +} + +#[cfg(test)] +mod rate_oracle { + use super::{BatchSummary, MetricsUpdate, StatsCollector, VpcMapName}; + use crate::PacketAndByte; + use crate::vpc_stats::VpcStatsStore; + use net::vxlan::Vni; + use vpcmap::VpcDiscriminant; + use vpcmap::map::VpcMapWriter; + + const LOAD: u64 = 1_000; + const SIZE: u64 = 500; + const CLOSE_ENOUGH: f64 = 1e-6; + + fn vpc(vni: u32) -> VpcDiscriminant { + VpcDiscriminant::from_vni(Vni::new_checked(vni).unwrap_or_else(|_| unreachable!())) + } + + fn a_tick_of(src: VpcDiscriminant, dst: VpcDiscriminant, packets: u64) -> MetricsUpdate { + let start = clock::now(); + let mut summary = BatchSummary::::new(start + StatsCollector::TIME_TICK); + summary.start = start; + let mut transmit = crate::TransmitSummary::::new(); + transmit.dst.insert( + dst, + PacketAndByte { + packets, + bytes: packets * SIZE, + }, + ); + summary.vpc.insert(src, transmit); + MetricsUpdate { + duration: StatsCollector::TIME_TICK, + summary: Box::new(summary), + } + } + + async fn published(ticks: usize, packets: u64) -> Option<(f64, f64)> { + let (src, dst) = (vpc(100), vpc(200)); + let (mut collector, store, _map) = collecting(src, dst); + for _ in 0..ticks { + collector.update(Some(a_tick_of(src, dst, packets))).await; + clock::virtual_time::advance(StatsCollector::TIME_TICK).await; + } + rate_of(&store, src, dst).await + } + + fn collecting( + src: VpcDiscriminant, + dst: VpcDiscriminant, + ) -> ( + StatsCollector, + concurrency::sync::Arc, + VpcMapWriter, + ) { + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "right"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + let (collector, _writer, store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + (collector, store, map) + } + + async fn rate_of( + store: &VpcStatsStore, + src: VpcDiscriminant, + dst: VpcDiscriminant, + ) -> Option<(f64, f64)> { + store + .snapshot_pairs() + .await + .into_iter() + .find(|((from, to), _)| *from == src && *to == dst) + .map(|(_, stats)| (stats.rate.pps, stats.rate.bps)) + } + + fn at_tick(t: usize) -> u64 { + LOAD + 100 * t as u64 + } + + async fn published_ramp(ticks: usize) -> Option<(f64, f64)> { + let (src, dst) = (vpc(100), vpc(200)); + let (mut collector, store, _map) = collecting(src, dst); + for t in 0..ticks { + collector + .update(Some(a_tick_of(src, dst, at_tick(t)))) + .await; + clock::virtual_time::advance(StatsCollector::TIME_TICK).await; + } + rate_of(&store, src, dst).await + } + + #[test] + fn a_steady_load_is_published_as_itself() { + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let published = published(24, LOAD).await; + let (pps, bps) = published.unwrap_or_else(|| { + panic!("the collector published no rate at all for a pair carrying {LOAD} pkt/s") + }); + assert!( + (pps - LOAD as f64).abs() < CLOSE_ENOUGH, + "{LOAD} pkt/s in, {pps} pkt/s out" + ); + let expect = (LOAD * SIZE) as f64; + assert!( + (bps - expect).abs() < CLOSE_ENOUGH, + "{expect} B/s in, {bps} B/s out" + ); + }); + } + + #[test] + fn a_steady_load_is_published_as_itself_at_every_tick() { + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + for ticks in 16..=32 { + let Some((pps, _)) = published(ticks, LOAD).await else { + panic!("no rate published after {ticks} ticks of {LOAD} pkt/s"); + }; + assert!( + (pps - LOAD as f64).abs() < CLOSE_ENOUGH, + "after {ticks} ticks, {LOAD} pkt/s in and {pps} pkt/s out" + ); + } + }); + } + + #[test] + fn a_changing_load_is_published_in_the_order_it_happened() { + const SETTLED: usize = 5; + const WARM: usize = SETTLED * 2; + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + for ticks in WARM..=(WARM + 16) { + let Some((pps, bps)) = published_ramp(ticks).await else { + panic!("no rate published after {ticks} ticks of a rising load"); + }; + let expect = at_tick(ticks - SETTLED) as f64; + assert!( + (pps - expect).abs() < CLOSE_ENOUGH, + "after {ticks} ticks of a rising load the published rate was {pps} pkt/s, \ + not the {expect} pkt/s offered {SETTLED} ticks ago" + ); + assert!( + (bps - expect * SIZE as f64).abs() < CLOSE_ENOUGH, + "the byte rate disagreed with the packet rate: {bps} B/s against {pps} pkt/s" + ); + } + }); + } + + #[test] + fn a_new_destination_does_not_silence_the_others() { + const ARRIVES: usize = 14; + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (src, old, new) = (vpc(100), vpc(200), vpc(300)); + let mut map = VpcMapWriter::::new(); + for (disc, name) in [(src, "left"), (old, "right"), (new, "newcomer")] { + map.add(disc, VpcMapName::new(disc, name), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + } + let (mut collector, store, _writer) = { + let (collector, _w, store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + (collector, store, _w) + }; + + for tick in 0..(ARRIVES + 12) { + let mut update = a_tick_of(src, old, LOAD); + if tick >= ARRIVES { + update + .summary + .vpc + .get_mut(&src) + .unwrap_or_else(|| unreachable!("the update names its source")) + .dst + .insert( + new, + PacketAndByte { + packets: LOAD, + bytes: LOAD * SIZE, + }, + ); + } + collector.update(Some(update)).await; + clock::virtual_time::advance(StatsCollector::TIME_TICK).await; + + let Some((pps, _)) = rate_of(&store, src, old).await else { + if tick < 10 { + continue; + } + panic!( + "at tick {tick} the established pair had no rate at all, and the only \ + thing that changed was another pair starting up" + ); + }; + if tick >= 10 { + assert!( + (pps - LOAD as f64).abs() < CLOSE_ENOUGH, + "at tick {tick} the established pair read {pps} pkt/s instead of {LOAD}" + ); + } + } + }); + } + + #[test] + fn every_packet_fed_in_is_eventually_counted() { + const TICKS: usize = 20; + const DRAIN: usize = 16; + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (src, dst) = (vpc(100), vpc(200)); + let (mut collector, store, _map) = collecting(src, dst); + for _ in 0..TICKS { + collector.update(Some(a_tick_of(src, dst, LOAD))).await; + clock::virtual_time::advance(StatsCollector::TIME_TICK).await; + } + for _ in 0..DRAIN { + collector.update(None).await; + clock::virtual_time::advance(StatsCollector::TIME_TICK).await; + } + let counted = store + .snapshot_pairs() + .await + .into_iter() + .find(|((from, to), _)| *from == src && *to == dst) + .map_or((0, 0), |(_, stats)| (stats.ctr.packets, stats.ctr.bytes)); + let fed = LOAD * TICKS as u64; + assert_eq!( + counted.0, fed, + "{fed} packets were fed and {} were counted after the pipeline drained", + counted.0 + ); + assert_eq!( + counted.1, + fed * SIZE, + "the byte total disagreed with the packets" + ); + }); + } + + #[test] + fn the_ledger_balances_at_every_tick() { + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (src, dst) = (vpc(100), vpc(200)); + let (mut collector, store, _map) = collecting(src, dst); + for tick in 0..24u64 { + collector.update(Some(a_tick_of(src, dst, LOAD))).await; + let held: u64 = collector + .outstanding + .iter() + .flat_map(|batch| batch.vpc.values()) + .flat_map(|summary| summary.dst.iter().map(|(_, v)| v.packets)) + .sum(); + let credited = store + .snapshot_pairs() + .await + .into_iter() + .find(|((from, to), _)| *from == src && *to == dst) + .map_or(0, |(_, stats)| stats.ctr.packets); + let fed = LOAD * (tick + 1); + assert_eq!( + credited + held, + fed, + "after {} ticks, {fed} packets had been fed but {credited} were counted and \ + {held} were still in open batches", + tick + 1 + ); + clock::virtual_time::advance(StatsCollector::TIME_TICK).await; + } + }); + } + + #[test] + fn the_open_batches_tile_the_time_ahead() { + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (collector, _store, _map) = collecting(vpc(100), vpc(200)); + let batches: Vec<_> = collector.outstanding.iter().collect(); + assert_eq!(batches.len(), StatsCollector::OUTSTANDING); + for (nth, batch) in batches.iter().enumerate() { + assert_eq!( + batch.planned_end.saturating_duration_since(batch.start), + StatsCollector::TIME_TICK, + "batch {nth} does not cover one tick" + ); + if let Some(prior) = nth.checked_sub(1) { + assert_eq!( + batch.start, batches[prior].planned_end, + "batch {nth} does not begin where batch {prior} ends" + ); + } + } + }); + } + + #[test] + fn the_ledger_balances_however_the_traffic_arrives() { + bolero::check!() + .with_type() + .cloned() + .for_each(|(loads, skews): (Vec, Vec)| { + if loads.is_empty() { + return; + } + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (src, dst) = (vpc(100), vpc(200)); + let (mut collector, store, _map) = collecting(src, dst); + let mut fed = 0u64; + for (nth, &load) in loads.iter().enumerate() { + let load = u64::from(load); + let skew = skews.get(nth).map_or(0, |&s| u64::from(s % 12)); + let mut update = a_tick_of(src, dst, load); + update.summary.start -= StatsCollector::TIME_TICK * skew as u32; + collector.update(Some(update)).await; + fed += load; + + let held: u64 = collector + .outstanding + .iter() + .flat_map(|batch| batch.vpc.values()) + .flat_map(|summary| summary.dst.iter().map(|(_, v)| v.packets)) + .sum(); + let credited = store + .snapshot_pairs() + .await + .into_iter() + .find(|((from, to), _)| *from == src && *to == dst) + .map_or(0, |(_, stats)| stats.ctr.packets); + assert_eq!( + credited + held, + fed, + "after {} updates, {fed} packets had been fed but {credited} were \ + counted and {held} were still in open batches", + nth + 1 + ); + clock::virtual_time::advance(StatsCollector::TIME_TICK).await; + } + }); + }); + } + + #[test] + fn an_idle_pair_is_published_as_idle() { + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let published = published(24, 0).await; + if let Some((pps, bps)) = published { + assert!( + pps.abs() < CLOSE_ENOUGH, + "an idle pair reported {pps} pkt/s" + ); + assert!(bps.abs() < CLOSE_ENOUGH, "an idle pair reported {bps} B/s"); + } + }); + } +} + +#[cfg(test)] +mod exported { + use super::{BatchSummary, MetricsUpdate, StatsCollector, VpcMapName}; + use crate::PacketAndByte; + use crate::scrape::Scrape; + use crate::vpc_stats::VpcStatsStore; + use net::vxlan::Vni; + use vpcmap::VpcDiscriminant; + use vpcmap::map::VpcMapWriter; + + const LOAD: u64 = 1_000; + const SIZE: u64 = 500; + const CLOSE_ENOUGH: f64 = 1e-6; + + fn vpc(vni: u32) -> VpcDiscriminant { + VpcDiscriminant::from_vni(Vni::new_checked(vni).unwrap_or_else(|_| unreachable!())) + } + + async fn traffic( + collector: &mut StatsCollector, + src: VpcDiscriminant, + dst: VpcDiscriminant, + ticks: usize, + ) { + for _ in 0..ticks { + let start = clock::now(); + let mut summary = BatchSummary::::new(start + StatsCollector::TIME_TICK); + summary.start = start; + let mut transmit = crate::TransmitSummary::::new(); + transmit.dst.insert( + dst, + PacketAndByte { + packets: LOAD, + bytes: LOAD * SIZE, + }, + ); + summary.vpc.insert(src, transmit); + collector + .update(Some(MetricsUpdate { + duration: StatsCollector::TIME_TICK, + summary: Box::new(summary), + })) + .await; + clock::virtual_time::advance(StatsCollector::TIME_TICK).await; + } + } + + async fn quiet(collector: &mut StatsCollector, ticks: usize) { + for _ in 0..ticks { + collector.update(None).await; + clock::virtual_time::advance(StatsCollector::TIME_TICK).await; + } + } + + fn pair_rates_leaving(scrape: &Scrape, from: &str) -> Vec<(String, f64)> { + scrape + .series("vpc_packet_rate") + .into_iter() + .filter(|(labels, rate)| { + *rate != 0.0 && labels.get("from").is_some_and(|name| name == from) + }) + .filter_map(|(labels, rate)| Some((labels.get("to")?.clone(), rate))) + .collect() + } + + #[test] + fn a_steady_load_reaches_the_gauge_an_operator_reads() { + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst) = (vpc(100), vpc(200)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "right"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, _store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, 24).await; + + let rate = scrape + .get("vpc_packet_rate", &[("from", "left"), ("to", "right")]) + .unwrap_or_else(|| { + panic!( + "no vpc_packet_rate series exists for left->right; exported:\n{}", + scrape.nonzero().join("\n") + ) + }); + assert!( + (rate - LOAD as f64).abs() < CLOSE_ENOUGH, + "{LOAD} pkt/s offered, {rate} pkt/s exported" + ); + }); + } + + #[test] + fn carrying_traffic_does_not_re_register_the_metrics() { + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst) = (vpc(100), vpc(200)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "right"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, _store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, 4).await; + let settled = scrape.registrations(); + traffic(&mut collector, src, dst, 40).await; + let after = scrape.registrations(); + assert_eq!( + settled, + after, + "40 further ticks of unchanged configuration cost {} metric registrations", + after - settled + ); + }); + } + + #[test] + fn a_configuration_change_does_re_register_the_metrics() { + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst) = (vpc(100), vpc(200)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "right"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, _store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, 4).await; + let settled = scrape.registrations(); + + map.add(vpc(300), VpcMapName::new(vpc(300), "third"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + traffic(&mut collector, src, dst, 4).await; + assert!( + scrape.registrations() > settled, + "a VPC was added and no new series was registered" + ); + }); + } + + #[test] + fn an_idle_peering_is_not_published() { + const FABRIC: u32 = 8; + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let mut map = VpcMapWriter::::new(); + for i in 0..FABRIC { + let disc = vpc(100 + i); + map.add( + disc, + VpcMapName::new(disc, &format!("vpc{i}")), + i + 1 == FABRIC, + ) + .unwrap_or_else(|e| unreachable!("{e:?}")); + } + let (src, dst) = (vpc(100), vpc(101)); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, 24).await; + + let held = store.snapshot_pairs().await; + assert_eq!( + held.len(), + 1, + "one peering out of {} carried traffic, but the store holds {}: {:?}", + FABRIC * FABRIC, + held.len(), + held.iter().map(|&(pair, _)| pair).collect::>() + ); + + let rate = scrape + .get("vpc_packet_rate", &[("from", "vpc0"), ("to", "vpc1")]) + .unwrap_or_else(|| unreachable!("the live pair was never exported")); + assert!( + (rate - LOAD as f64).abs() < CLOSE_ENOUGH, + "{LOAD} pkt/s offered, {rate} pkt/s exported" + ); + }); + } + + #[test] + fn a_peering_that_falls_idle_is_published_as_idle() { + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst) = (vpc(100), vpc(200)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "right"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, _store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, 24).await; + quiet(&mut collector, 16).await; + + let rate = scrape + .get("vpc_packet_rate", &[("from", "left"), ("to", "right")]) + .unwrap_or_else(|| unreachable!("the pair was never exported at all")); + assert!( + rate.abs() < CLOSE_ENOUGH, + "the link stopped carrying traffic but still reports {rate} pkt/s" + ); + }); + } + + #[test] + fn a_recycled_vni_does_not_inherit_the_previous_tenants_counters() { + const BEFORE: usize = 24; + const AFTER: usize = 6; + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst) = (vpc(100), vpc(200)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "customer-a"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, _store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, BEFORE).await; + + map.add(dst, VpcMapName::new(dst, "customer-b"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + traffic(&mut collector, src, dst, AFTER).await; + + let counted = scrape + .get( + "vpc_packet_count", + &[("from", "left"), ("to", "customer-b")], + ) + .unwrap_or_else(|| unreachable!("the new tenant was never exported")); + assert!( + counted <= (AFTER as u64 * LOAD) as f64, + "customer-b has sent at most {} packets but is credited with {counted}; \ + customer-a sent {}", + AFTER as u64 * LOAD, + BEFORE as u64 * LOAD + ); + assert!( + scrape + .get( + "vpc_packet_count", + &[("from", "left"), ("to", "customer-a")] + ) + .is_none_or(|stale| stale.abs() < CLOSE_ENOUGH), + "customer-a's series is still exporting after the VNI was handed on" + ); + }); + } + + #[test] + fn every_series_has_a_label_shape_the_family_allows() { + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst) = (vpc(100), vpc(200)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "right"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, _store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, 4).await; + map.add(vpc(300), VpcMapName::new(vpc(300), "third"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + traffic(&mut collector, src, dst, 4).await; + + let allowed: std::collections::BTreeSet> = [ + vec!["total".to_string()], + vec!["drops".to_string()], + vec!["from".to_string(), "to".to_string()], + ] + .into_iter() + .collect(); + for family in ["vpc_packet_count", "vpc_packet_rate", "vpc_byte_count"] { + let found = scrape.label_shapes(family); + assert!( + found.is_subset(&allowed), + "{family} is exported with label shapes {:?}", + found.difference(&allowed).collect::>() + ); + } + }); + } + + #[test] + fn a_recycled_vni_is_not_credited_with_traffic_in_flight() { + const DRAIN: usize = 14; + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst) = (vpc(100), vpc(200)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "customer-a"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, 24).await; + + map.add(dst, VpcMapName::new(dst, "customer-b"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + for tick in 0..DRAIN { + quiet(&mut collector, 1).await; + let credited = store + .snapshot_pairs() + .await + .into_iter() + .find(|&((from, to), _)| from == src && to == dst) + .map_or(0, |(_, stats)| stats.ctr.packets); + assert_eq!( + credited, 0, + "customer-b has sent nothing, and {tick} ticks after taking the VNI over it \ + is credited with {credited} packets" + ); + let rate = scrape + .get( + "vpc_packet_count", + &[("from", "left"), ("to", "customer-b")], + ) + .unwrap_or(0.0); + assert!( + rate.abs() < CLOSE_ENOUGH, + "customer-b has sent nothing but its count gauge reads {rate}" + ); + } + }); + } + + #[test] + fn deleting_a_peer_does_not_subtract_from_a_survivors_total() { + const TICKS: u64 = 24; + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst) = (vpc(100), vpc(200)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "right"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, _store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, TICKS as usize).await; + + map.del(dst, true); + quiet(&mut collector, 16).await; + + let offered = (TICKS * LOAD) as f64; + let counted = scrape + .get("vpc_packet_count", &[("total", "left")]) + .unwrap_or_else(|| unreachable!("left has no total at all")); + assert!( + (counted - offered).abs() < CLOSE_ENOUGH, + "left sent {offered} packets and is credited with {counted} after its peer was \ + deleted" + ); + }); + } + + #[test] + fn a_map_that_cannot_be_read_is_not_read_as_empty() { + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst) = (vpc(100), vpc(200)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "right"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, 24).await; + let held = store.snapshot_pairs().await; + assert!(!held.is_empty(), "nothing was counted to begin with"); + + drop(map); + quiet(&mut collector, 4).await; + + assert_eq!( + store.snapshot_pairs().await.len(), + held.len(), + "the counters were discarded because the map could not be read" + ); + assert!( + scrape + .get("vpc_packet_count", &[("from", "left"), ("to", "right")]) + .is_some_and(|counted| counted > 0.0), + "every series was zeroed because the map could not be read" + ); + }); + } + + #[test] + fn renaming_a_vpc_does_not_leave_its_old_series_running() { + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst) = (vpc(100), vpc(200)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "right"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, _store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, 24).await; + + map.add(dst, VpcMapName::new(dst, "renamed"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + traffic(&mut collector, src, dst, 12).await; + + let live = pair_rates_leaving(&scrape, "left"); + let total: f64 = live.iter().map(|(_, rate)| rate).sum(); + assert!( + (total - LOAD as f64).abs() < CLOSE_ENOUGH, + "one link carrying {LOAD} pkt/s is exported as {total} pkt/s across {live:?}; \ + exported:\n{}", + scrape.nonzero().join("\n") + ); + }); + } + + #[test] + fn a_peering_whose_ends_have_both_gone_stops_being_exported() { + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst, bystander) = (vpc(100), vpc(200), vpc(300)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "right"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(bystander, VpcMapName::new(bystander, "bystander"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, _store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, 24).await; + + map.del(src, false); + map.del(dst, true); + quiet(&mut collector, 8).await; + + let stale = pair_rates_leaving(&scrape, "left"); + assert!( + stale.is_empty(), + "both ends of left->right were deleted, but {stale:?} is still exported; \ + exported:\n{}", + scrape.nonzero().join("\n") + ); + }); + } + + #[test] + fn a_peering_whose_far_end_has_gone_stops_being_exported() { + let scrape = Scrape::default(); + let _installed = metrics::set_default_local_recorder(&scrape); + let (src, dst) = (vpc(100), vpc(200)); + let mut map = VpcMapWriter::::new(); + map.add(src, VpcMapName::new(src, "left"), false) + .unwrap_or_else(|e| unreachable!("{e:?}")); + map.add(dst, VpcMapName::new(dst, "right"), true) + .unwrap_or_else(|e| unreachable!("{e:?}")); + + let clock = clock::virtual_time::Paused::new(); + clock.block_on(async { + let (mut collector, _writer, _store) = + StatsCollector::new_with_store(map.get_reader(), VpcStatsStore::new()); + traffic(&mut collector, src, dst, 24).await; + + map.del(dst, true); + quiet(&mut collector, 8).await; + + let stale = pair_rates_leaving(&scrape, "left"); + assert!( + stale.is_empty(), + "right was deleted, but {stale:?} is still exported; exported:\n{}", + scrape.nonzero().join("\n") + ); + }); } } diff --git a/stats/src/lib.rs b/stats/src/lib.rs index 7e9bb5ea82..51a938b4bb 100644 --- a/stats/src/lib.rs +++ b/stats/src/lib.rs @@ -8,6 +8,8 @@ mod dpstats_fuzz; mod rate; mod rate_fuzz; mod register; +#[cfg(test)] +mod scrape; mod spec; mod vpc; mod vpc_stats; diff --git a/stats/src/rate.rs b/stats/src/rate.rs index 759cc6c334..60e734bcb8 100644 --- a/stats/src/rate.rs +++ b/stats/src/rate.rs @@ -88,11 +88,13 @@ pub trait HashMapSmoothing { /// f^{\prime}\!\left(x\right) \approx \frac{8 \left[f\!\left(x + h\right) - f\!\left(x - h\right)\right] - \left[f\!\left(x + 2h\right) - f\!\left(x - 2h\right)\right]}{12 h} /// } /// ``` +const WINDOW: usize = 5; + #[derive(Debug)] pub struct SavitzkyGolayFilter { step: Duration, idx: usize, - data: ArrayVec, + data: ArrayVec, } impl Default for SavitzkyGolayFilter { @@ -117,7 +119,19 @@ impl SavitzkyGolayFilter { self.data[self.idx] = e.element(); } } - self.idx = (self.idx + 1) % 5; + self.idx = (self.idx + 1) % WINDOW; + } + + pub fn chronological(&self) -> impl Iterator { + self.data + .iter() + .cycle() + .skip(self.idx) + .take(self.data.len()) + } + + pub fn each_sample_mut(&mut self, mut edit: impl FnMut(&mut U)) { + self.data.iter_mut().for_each(&mut edit); } } @@ -125,19 +139,21 @@ impl SavitzkyGolayFilter { pub enum DerivativeError { #[error("Not enough samples to compute derivative: {0} available")] NotEnoughSamples(usize), + #[error("A zero time step has no derivative: a rate over it would divide by zero")] + ZeroStep, } impl Derivative for SavitzkyGolayFilter { type Error = DerivativeError; type Output = f64; fn derivative(&self) -> Result { - const SAMPLES: usize = 5; + const SAMPLES: usize = WINDOW; let data_len = self.data.len(); if data_len < SAMPLES { return Err(DerivativeError::NotEnoughSamples(data_len)); } debug_assert!(data_len == SAMPLES); - let mut itr = self.data.iter().cycle().skip(self.idx).copied(); + let mut itr = self.chronological().copied(); let data: [u64; SAMPLES] = [ itr.next().unwrap_or_else(|| unreachable!()), itr.next().unwrap_or_else(|| unreachable!()), @@ -148,6 +164,9 @@ impl Derivative for SavitzkyGolayFilter { let weighted_sum = 8u64 .saturating_mul(data[3].saturating_sub(data[1])) .saturating_sub(data[4].saturating_sub(data[0])); + if self.step.as_micros() == 0 { + return Err(DerivativeError::ZeroStep); + } let step: f64 = self.step.as_micros() as f64 / 1_000_000.; if weighted_sum == 0 { const NORMALIZATION: f64 = 2.; @@ -162,12 +181,12 @@ impl Derivative for SavitzkyGolayFilter> { type Error = DerivativeError; type Output = PacketAndByte; fn derivative(&self) -> Result, DerivativeError> { - const SAMPLES: usize = 5; + const SAMPLES: usize = WINDOW; let data_len = self.data.len(); if data_len < SAMPLES { return Err(DerivativeError::NotEnoughSamples(data_len)); } - let mut itr = self.data.iter().cycle().skip(self.idx).copied(); + let mut itr = self.chronological().copied(); let data: [PacketAndByte; SAMPLES] = [ itr.next().unwrap_or_else(|| unreachable!()), itr.next().unwrap_or_else(|| unreachable!()), @@ -176,8 +195,11 @@ impl Derivative for SavitzkyGolayFilter> { itr.next().unwrap_or_else(|| unreachable!()), ]; let weighted_sum_bytes = 8u64 - .saturating_mul(data[3].bytes - data[1].bytes) - .saturating_sub(data[4].bytes - data[0].bytes); + .saturating_mul(data[3].bytes.saturating_sub(data[1].bytes)) + .saturating_sub(data[4].bytes.saturating_sub(data[0].bytes)); + if self.step.as_micros() == 0 { + return Err(DerivativeError::ZeroStep); + } let step: f64 = self.step.as_micros() as f64 / 1_000_000.; if weighted_sum_bytes == 0 { const NORMALIZATION: f64 = 2.; @@ -203,46 +225,32 @@ impl TryFrom<&SavitzkyGolayFilter>> type Error = DerivativeError; fn try_from(value: &SavitzkyGolayFilter>) -> Result { - if value.data.len() != 5 { + if value.data.len() != WINDOW { return Err(DerivativeError::NotEnoughSamples(value.data.len())); } - let values: Vec<_> = value - .data - .iter() - .cycle() - .skip(value.idx) - .take(5) - .cloned() - .collect(); - let all_keys: BTreeSet<_> = values - .iter() + let all_keys: BTreeSet<_> = value + .chronological() .flat_map(|x| x.dst.iter().map(|(&k, _)| k)) .collect(); let mut out = TransmitSummary::>::new(); - for (idx, summary) in values.iter().enumerate() { - all_keys - .iter() - .for_each(|&k| match (summary.dst.get(&k), out.dst.get_mut(&k)) { - (Some(count), Some(out)) => { - out.packets.push(count.packets); - out.bytes.push(count.bytes); - } - (Some(count), None) => { - let mut packets = SavitzkyGolayFilter::new(value.step); - let mut bytes = SavitzkyGolayFilter::new(value.step); - packets.push(count.packets); - bytes.push(count.bytes); - out.dst.insert(k, PacketAndByte { packets, bytes }); - } - (None, Some(out)) => { - debug_assert!(idx != 0); - out.packets.push(out.packets.data[out.packets.idx - 1]); - out.bytes.push(out.bytes.data[out.bytes.idx - 1]); - } - (None, None) => { - // no data yet - } - }); + for &k in &all_keys { + out.dst.insert( + k, + PacketAndByte { + packets: SavitzkyGolayFilter::new(value.step), + bytes: SavitzkyGolayFilter::new(value.step), + }, + ); + } + for summary in value.chronological() { + for &k in &all_keys { + let Some(out) = out.dst.get_mut(&k) else { + unreachable!("every key was inserted above") + }; + let count = summary.dst.get(&k).copied().unwrap_or_default(); + out.packets.push(count.packets); + out.bytes.push(count.bytes); + } } Ok(out) } @@ -253,7 +261,7 @@ impl Derivative for SavitzkyGolayFilter> { type Output = TransmitSummary; fn derivative(&self) -> Result { - if self.data.len() != 5 { + if self.data.len() != WINDOW { return Err(DerivativeError::NotEnoughSamples(self.data.len())); } let x = TransmitSummary::>::try_from(self)?; @@ -349,23 +357,11 @@ impl From { - unreachable!(); // all keys in map should already be here - } - Some(filter) => { - filter.push(from.clone()); - } - }) + value.chronological().for_each(|map| { + out.iter_mut().for_each(|(key, filter)| { + filter.push(map.get(key).cloned().unwrap_or_default()); }); + }); out } } @@ -378,47 +374,51 @@ impl From<&SavitzkyGolayFilter Self { const CAPACITY_PAD: usize = 32; let capacity_guess = value.data.iter().map(|map| map.len()).max().unwrap_or(0); - let mut out = hashbrown::HashMap::with_capacity(capacity_guess + CAPACITY_PAD); + let mut pairs: hashbrown::HashMap> = + hashbrown::HashMap::with_capacity(capacity_guess + CAPACITY_PAD); value.data.iter().for_each(|map| { - map.iter().for_each(|(k, _)| { - if out.get(k).is_none() { - out.insert(*k, TransmitSummary::>::new()); - } + map.iter().for_each(|(&src, summary)| { + let seen = pairs.entry(src).or_default(); + summary.dst.iter().for_each(|(&dst, _)| { + seen.insert(dst); + }); }) }); - value.data.iter().enumerate().for_each(|(idx, map)| { - map.iter() - .for_each(|(from_key, from)| match out.get_mut(from_key) { - None => { - unreachable!(); // all keys in map should already be here - } - Some(summary) => { - from.dst.iter().for_each(|(to_key, to)| { - match summary.dst.get_mut(to_key) { - None => { - let mut packets = SavitzkyGolayFilter::new(value.step); - let mut bytes = SavitzkyGolayFilter::new(value.step); - packets.push(to.packets); - bytes.push(to.bytes); - - summary - .dst - .insert(*to_key, PacketAndByte { packets, bytes }); - } - Some(x) => { - while x.packets.idx < idx { - x.packets.push(x.packets.data[x.packets.idx - 1]); - } - while x.bytes.idx < idx { - x.bytes.push(x.bytes.data[x.bytes.idx - 1]); - } - x.packets.push(to.packets); - x.bytes.push(to.bytes); - } - } - }); - } - }) + let mut out: hashbrown::HashMap< + VpcDiscriminant, + TransmitSummary>, + > = hashbrown::HashMap::with_capacity(pairs.len() + CAPACITY_PAD); + for (&src, dsts) in &pairs { + let mut summary = TransmitSummary::>::new(); + for &dst in dsts { + summary.dst.insert( + dst, + PacketAndByte { + packets: SavitzkyGolayFilter::new(value.step), + bytes: SavitzkyGolayFilter::new(value.step), + }, + ); + } + out.insert(src, summary); + } + value.chronological().for_each(|map| { + for (&src, dsts) in &pairs { + let Some(summary) = out.get_mut(&src) else { + unreachable!("every source was inserted above") + }; + let observed = map.get(&src); + for &dst in dsts { + let Some(filter) = summary.dst.get_mut(&dst) else { + unreachable!("every destination was inserted above") + }; + let count = observed + .and_then(|summary| summary.dst.get(&dst)) + .copied() + .unwrap_or_default(); + filter.packets.push(count.packets); + filter.bytes.push(count.bytes); + } + } }); out } @@ -483,7 +483,7 @@ impl Smooth for SavitzkyGolayFilter { type Output = f64; fn smooth(&self) -> Result { - const SAMPLES: usize = 5; + const SAMPLES: usize = WINDOW; const COEFFS: [i64; SAMPLES] = [-3, 12, 17, 12, -3]; // / 35 const DEN: f64 = 35.0; @@ -493,7 +493,7 @@ impl Smooth for SavitzkyGolayFilter { } debug_assert!(len == SAMPLES); - let mut itr = self.data.iter().cycle().skip(self.idx).copied(); + let mut itr = self.chronological().copied(); let data: [u64; SAMPLES] = [ itr.next().unwrap_or_else(|| unreachable!()), itr.next().unwrap_or_else(|| unreachable!()), @@ -508,7 +508,7 @@ impl Smooth for SavitzkyGolayFilter { .zip(data.iter()) .fold(0i128, |s, (&c, &v)| s + (c as i128) * (v as i128)); - Ok((acc as f64) / DEN) + Ok(((acc as f64) / DEN).max(0.0)) } } @@ -517,7 +517,7 @@ impl Smooth for SavitzkyGolayFilter> { type Output = PacketAndByte; fn smooth(&self) -> Result { - const SAMPLES: usize = 5; + const SAMPLES: usize = WINDOW; const COEFFS: [i64; SAMPLES] = [-3, 12, 17, 12, -3]; // / 35 const DEN: f64 = 35.0; @@ -526,7 +526,7 @@ impl Smooth for SavitzkyGolayFilter> { return Err(DerivativeError::NotEnoughSamples(len)); } - let mut itr = self.data.iter().cycle().skip(self.idx).copied(); + let mut itr = self.chronological().copied(); let data: [PacketAndByte; SAMPLES] = [ itr.next().unwrap_or_else(|| unreachable!()), itr.next().unwrap_or_else(|| unreachable!()), @@ -545,8 +545,8 @@ impl Smooth for SavitzkyGolayFilter> { .fold(0i128, |s, (&c, v)| s + (c as i128) * (v.bytes as i128)); Ok(PacketAndByte { - packets: (acc_packets as f64) / DEN, - bytes: (acc_bytes as f64) / DEN, + packets: ((acc_packets as f64) / DEN).max(0.0), + bytes: ((acc_bytes as f64) / DEN).max(0.0), }) } } @@ -556,7 +556,7 @@ impl Smooth for SavitzkyGolayFilter> { type Output = TransmitSummary; fn smooth(&self) -> Result { - if self.data.len() != 5 { + if self.data.len() != WINDOW { return Err(DerivativeError::NotEnoughSamples(self.data.len())); } // Convert to per-destination SG filters first, then smooth those. @@ -613,7 +613,7 @@ where mod contract { use crate::rate::{Derivative, SavitzkyGolayFilter}; use crate::{PacketAndByte, TransmitSummary}; - use bolero::{Driver, TypeGenerator}; + use bolero::{Driver, TypeGenerator, ValueGenerator}; use std::fmt::Debug; use std::time::Duration; @@ -680,10 +680,74 @@ mod contract { } } + pub struct PerIntervalCounts; + + impl ValueGenerator for PerIntervalCounts { + type Output = SavitzkyGolayFilter; + + fn generate(&self, driver: &mut D) -> Option { + let mut step: Duration = driver.produce()?; + if step == Duration::ZERO { + step += Duration::from_secs(1); + } + let mut filter = SavitzkyGolayFilter::new(step); + let entries: u8 = driver.produce::()? % 15; + let ceiling = 1u64 << (driver.produce::()? % 63); + for _ in 0..entries { + filter.push(driver.produce::()? % ceiling); + } + Some(filter) + } + } + + pub struct PerIntervalPacketAndByte; + + impl ValueGenerator for PerIntervalPacketAndByte { + type Output = SavitzkyGolayFilter>; + + fn generate(&self, driver: &mut D) -> Option { + let mut step: Duration = driver.produce()?; + if step == Duration::ZERO { + step += Duration::from_secs(1); + } + let mut filter = SavitzkyGolayFilter::new(step); + let entries: u8 = driver.produce::()? % 15; + let ceiling = 1u64 << (driver.produce::()? % 63); + for _ in 0..entries { + filter.push(PacketAndByte { + packets: driver.produce::()? % ceiling, + bytes: driver.produce::()? % ceiling, + }); + } + Some(filter) + } + } + + pub mod naive { + use std::time::Duration; + + pub fn seconds(step: Duration) -> f64 { + step.as_micros() as f64 / 1_000_000. + } + + pub fn two_point(window: &[u64; 5], step: Duration) -> f64 { + (window[4] as f64 - window[3] as f64) / seconds(step) + } + + pub fn central(window: &[u64; 5], step: Duration) -> f64 { + (window[3] as f64 - window[1] as f64) / (2. * seconds(step)) + } + + pub fn mean(window: &[u64; 5]) -> f64 { + window.iter().map(|&v| v as f64).sum::() / 5. + } + } + pub struct DerivativeComparer { pub f: F, pub d: D, pub step: Duration, + pub carried: usize, } impl DerivativeComparer @@ -703,6 +767,9 @@ mod contract { x: Duration, ) -> DerivativeComparison< as Derivative>::Output> { let mut out = SavitzkyGolayFilter::new(self.step); + for i in 0..self.carried { + out.push((self.f)(x + self.step * u32::try_from(i).unwrap())); + } for i in 0..5 { out.push((self.f)(x + self.step * u32::try_from(i).unwrap())); } @@ -739,6 +806,8 @@ mod contract { #[cfg(test)] mod test { use crate::rate::{Derivative, DerivativeComparer, DerivativeError, SavitzkyGolayFilter}; + use concurrency::sync::LazyLock; + use concurrency::sync::atomic::{AtomicU64, Ordering}; use crate::{PacketAndByte, TransmitSummary}; @@ -749,10 +818,8 @@ mod test { macro_rules! arbitrary_polynomial { ($n:expr) => {{ const NANOS_PER_SEC: u128 = 1_000_000_000; - bolero::check!() - .with_type() - .cloned() - .for_each(|(x, c): (Duration, [u64; $n])| { + bolero::check!().with_type().cloned().for_each( + |(x, c, carried): (Duration, [u64; $n], u8)| { let x = if x < Duration::from_micros(1) { Duration::from_micros(1) } else if x > Duration::from_secs(10) { @@ -783,6 +850,7 @@ mod test { f: basic, d: basic_prime, step: Duration::from_secs(1), + carried: usize::from(carried) % 5, }; let comparison = comparer.compare(x); if comparison.relative_error().is_nan() { @@ -790,7 +858,8 @@ mod test { return; } assert!(comparison.relative_error().abs() < 0.01); - }) + }, + ) }}; } #[test] @@ -844,18 +913,82 @@ mod test { arbitrary_polynomial!(12); } + #[test] + fn smoothing_a_counter_is_never_negative() { + static UNDERSHOT: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + bolero::check!() + .with_generator(crate::rate::contract::PerIntervalCounts) + .for_each(|x: &SavitzkyGolayFilter| { + let Ok(v) = x.smooth() else { + return; + }; + assert!(v >= 0.0, "smoothed a counter to {v}"); + if unclamped(&x.chronological().copied().collect::>()) < 0. { + UNDERSHOT.fetch_add(1, Ordering::Relaxed); + } + }); + + let undershot = UNDERSHOT.load(Ordering::Relaxed); + assert!( + undershot > 0, + "no window in the whole run would have smoothed to a negative number, so this said \ + nothing about the clamp it exists to guard" + ); + } + + #[test] + fn smoothing_a_packet_and_byte_counter_is_never_negative() { + static UNDERSHOT: LazyLock = LazyLock::new(|| AtomicU64::new(0)); + + bolero::check!() + .with_generator(crate::rate::contract::PerIntervalPacketAndByte) + .for_each(|x: &SavitzkyGolayFilter>| { + let Ok(v) = x.smooth() else { + return; + }; + assert!(v.packets >= 0.0, "smoothed a packet count to {}", v.packets); + assert!(v.bytes >= 0.0, "smoothed a byte count to {}", v.bytes); + let window: Vec<_> = x.chronological().copied().collect(); + let packets: Vec = window.iter().map(|v| v.packets).collect(); + let bytes: Vec = window.iter().map(|v| v.bytes).collect(); + if unclamped(&packets) < 0. || unclamped(&bytes) < 0. { + UNDERSHOT.fetch_add(1, Ordering::Relaxed); + } + }); + + let undershot = UNDERSHOT.load(Ordering::Relaxed); + assert!( + undershot > 0, + "no window in the whole run would have smoothed to a negative number, so this said \ + nothing about the clamp it exists to guard" + ); + } + #[test] fn derivative_filter_basic() { bolero::check!() .with_type() .for_each(|x: &SavitzkyGolayFilter| match x.derivative() { - Ok(x) => { - assert!(x >= 0.0); + Ok(d) => { + assert!( + d >= 0.0, + "every operand is a saturating unsigned subtraction, so the only way to \ + fail this is NaN: got {d} at step {:?}", + x.step + ); } Err(DerivativeError::NotEnoughSamples(s)) => { assert_eq!(x.idx, s); assert!(s < 5); } + Err(DerivativeError::ZeroStep) => { + assert_eq!( + x.step.as_micros(), + 0, + "a step is refused only when it rounds to zero microseconds" + ); + } }) } @@ -865,15 +998,16 @@ mod test { .with_type() .for_each( |x: &SavitzkyGolayFilter>| match x.derivative() { - Ok(x) => { - if !x.packets.is_nan() { - assert!(x.packets >= 0.0); - assert!(x.bytes >= 0.0); - } + Ok(d) => { + assert!(d.packets >= 0.0, "{:?} at step {:?}", d, x.step); + assert!(d.bytes >= 0.0, "{:?} at step {:?}", d, x.step); } Err(DerivativeError::NotEnoughSamples(s)) => { assert_eq!(x.idx, s); } + Err(DerivativeError::ZeroStep) => { + assert_eq!(x.step.as_micros(), 0); + } }, ) } @@ -893,10 +1027,25 @@ mod test { Err(DerivativeError::NotEnoughSamples(s)) => { assert!(s < 5) } + Err(DerivativeError::ZeroStep) => { + assert_eq!(x.step.as_micros(), 0); + } }, ) } + fn unclamped(window: &[u64]) -> f64 { + const COEFFS: [i64; 5] = [-3, 12, 17, 12, -3]; + if window.len() < 5 { + return 0.; + } + COEFFS + .iter() + .zip(window.iter()) + .fold(0i128, |acc, (&c, &v)| acc + i128::from(c) * i128::from(v)) as f64 + / 35. + } + use crate::rate::Smooth; #[test] @@ -1043,3 +1192,321 @@ mod test { assert!((out.bytes - (440.0 / 35.0)).abs() < 1e-9); } } + +#[cfg(test)] +mod second_opinion { + use crate::rate::contract::naive; + use crate::rate::{Derivative, SavitzkyGolayFilter, Smooth, WINDOW}; + use std::time::Duration; + + const CLOSE_ENOUGH: f64 = 1e-9; + + fn wound(window: &[u64; WINDOW], rotation: usize, step: Duration) -> SavitzkyGolayFilter { + let mut filter = SavitzkyGolayFilter::new(step); + for i in 0..rotation { + filter.push(window[i % WINDOW]); + } + for value in window { + filter.push(*value); + } + filter + } + + fn usable(step: Duration) -> Duration { + step.clamp(Duration::from_micros(1), Duration::from_secs(60)) + } + + #[test] + fn a_flat_line_derives_to_nothing_however_the_ring_sits() { + bolero::check!().with_type().cloned().for_each( + |(level, step, rotation): (u32, Duration, u8)| { + let step = usable(step); + let window = [u64::from(level); WINDOW]; + let filter = wound(&window, usize::from(rotation) % WINDOW, step); + let got = filter.derivative().unwrap_or_else(|e| unreachable!("{e}")); + assert!(got.abs() < CLOSE_ENOUGH, "a flat line derived to {got}"); + assert!(naive::two_point(&window, step).abs() < CLOSE_ENOUGH); + assert!(naive::central(&window, step).abs() < CLOSE_ENOUGH); + }, + ); + } + + #[test] + fn a_straight_line_derives_to_its_slope_however_the_ring_sits() { + bolero::check!().with_type().cloned().for_each( + |(base, slope, step, rotation): (u32, u16, Duration, u8)| { + let step = usable(step); + let (base, slope) = (u64::from(base), u64::from(slope)); + let window: [u64; WINDOW] = std::array::from_fn(|i| base + slope * (i as u64)); + let filter = wound(&window, usize::from(rotation) % WINDOW, step); + let got = filter.derivative().unwrap_or_else(|e| unreachable!("{e}")); + let want = slope as f64 / naive::seconds(step); + let scale = want.abs().max(1.0); + assert!( + (got - want).abs() / scale < CLOSE_ENOUGH, + "a line rising {slope} per step derived to {got}, not {want}" + ); + assert!((got - naive::two_point(&window, step)).abs() / scale < CLOSE_ENOUGH); + assert!((got - naive::central(&window, step)).abs() / scale < CLOSE_ENOUGH); + }, + ); + } + + #[test] + fn a_flat_line_smooths_to_itself_however_the_ring_sits() { + bolero::check!().with_type().cloned().for_each( + |(level, step, rotation): (u32, Duration, u8)| { + let step = usable(step); + let window = [u64::from(level); WINDOW]; + let filter = wound(&window, usize::from(rotation) % WINDOW, step); + let got = filter.smooth().unwrap_or_else(|e| unreachable!("{e}")); + let want = f64::from(level); + let scale = want.abs().max(1.0); + assert!( + (got - want).abs() / scale < CLOSE_ENOUGH, + "a flat {level} smoothed to {got}" + ); + assert!((got - naive::mean(&window)).abs() / scale < CLOSE_ENOUGH); + }, + ); + } + + #[test] + fn a_straight_line_smooths_to_its_middle_sample_however_the_ring_sits() { + bolero::check!().with_type().cloned().for_each( + |(base, slope, step, rotation): (u32, u16, Duration, u8)| { + let step = usable(step); + let (base, slope) = (u64::from(base), u64::from(slope)); + let window: [u64; WINDOW] = std::array::from_fn(|i| base + slope * (i as u64)); + let filter = wound(&window, usize::from(rotation) % WINDOW, step); + let got = filter.smooth().unwrap_or_else(|e| unreachable!("{e}")); + let middle = window[WINDOW / 2] as f64; + let scale = middle.abs().max(1.0); + assert!( + (got - middle).abs() / scale < CLOSE_ENOUGH, + "a line through {middle} smoothed to {got}" + ); + assert!((got - naive::mean(&window)).abs() / scale < CLOSE_ENOUGH); + }, + ); + } +} + +#[cfg(test)] +mod window_order { + use crate::rate::{DerivativeError, SavitzkyGolayFilter, Smooth, WINDOW}; + use crate::{PacketAndByte, TransmitSummary}; + use std::time::Duration; + use vpcmap::VpcDiscriminant; + + fn vpc(n: u32) -> VpcDiscriminant { + VpcDiscriminant::from_vni(n.try_into().unwrap_or_else(|_| unreachable!())) + } + + fn rising_load( + count: usize, + ) -> SavitzkyGolayFilter>> { + let mut window = SavitzkyGolayFilter::new(Duration::from_secs(1)); + for tick in 0..count as u64 { + let mut summary = TransmitSummary::::new(); + summary.dst.insert( + vpc(2), + PacketAndByte { + packets: at_tick(tick), + bytes: at_tick(tick) * 100, + }, + ); + let mut by_src = hashbrown::HashMap::new(); + by_src.insert(vpc(1), summary); + window.push(by_src); + } + window + } + + const BASE: u64 = 1_000; + const SLOPE: u64 = 100; + + fn at_tick(tick: u64) -> u64 { + BASE + SLOPE * tick + } + + fn expected_after(ticks: usize) -> f64 { + at_tick(ticks as u64 - 3) as f64 + } + + #[test] + fn a_rising_load_reads_the_same_at_every_ring_offset() { + for ticks in WINDOW..=(4 * WINDOW) { + let window = rising_load(ticks); + let by_src: hashbrown::HashMap< + VpcDiscriminant, + TransmitSummary>, + > = (&window).into(); + let smoothed = by_src + .get(&vpc(1)) + .unwrap_or_else(|| unreachable!()) + .smooth() + .unwrap_or_else(|_| unreachable!()); + let rate = smoothed + .dst + .get(&vpc(2)) + .unwrap_or_else(|| unreachable!()) + .packets; + let expect = expected_after(ticks); + assert!( + (rate - expect).abs() < 1e-9, + "after {ticks} ticks the ramp smoothed to {rate} pkt/s, not {expect}" + ); + } + } + + #[test] + fn a_rising_load_derives_the_same_at_every_ring_offset() { + for ticks in WINDOW..=(4 * WINDOW) { + let window = rising_load(ticks); + let by_src: hashbrown::HashMap< + VpcDiscriminant, + SavitzkyGolayFilter>, + > = window.into(); + let smoothed = by_src + .get(&vpc(1)) + .unwrap_or_else(|| unreachable!()) + .smooth() + .unwrap_or_else(|_| unreachable!()); + let rate = smoothed + .dst + .get(&vpc(2)) + .unwrap_or_else(|| unreachable!()) + .packets; + let expect = expected_after(ticks); + assert!( + (rate - expect).abs() < 1e-9, + "after {ticks} ticks the ramp smoothed to {rate} pkt/s, not {expect}" + ); + } + } + + #[test] + fn a_destination_that_appears_late_does_not_hide_the_others() { + for first_seen in 0..WINDOW { + let mut window = SavitzkyGolayFilter::new(Duration::from_secs(1)); + for tick in 0..WINDOW { + let mut summary = TransmitSummary::::new(); + summary.dst.insert( + vpc(2), + PacketAndByte { + packets: 100, + bytes: 10_000, + }, + ); + if tick >= first_seen { + summary.dst.insert( + vpc(3), + PacketAndByte { + packets: 7, + bytes: 700, + }, + ); + } + let mut by_src = hashbrown::HashMap::new(); + by_src.insert(vpc(1), summary); + window.push(by_src); + } + let by_src: hashbrown::HashMap< + VpcDiscriminant, + TransmitSummary>, + > = (&window).into(); + let smoothed = by_src + .get(&vpc(1)) + .unwrap_or_else(|| unreachable!()) + .smooth() + .unwrap_or_else(|e| panic!("vpc 3 first seen at tick {first_seen}: {e}")); + let steady = smoothed + .dst + .get(&vpc(2)) + .unwrap_or_else(|| unreachable!()) + .packets; + assert!( + (steady - 100.0).abs() < 1e-9, + "a steady destination read {steady} pkt/s because another arrived at tick \ + {first_seen}" + ); + } + } + + #[test] + fn a_destination_that_stops_reads_as_idle() { + let mut window = SavitzkyGolayFilter::new(Duration::from_secs(1)); + for tick in 0..WINDOW { + let mut summary = TransmitSummary::::new(); + if tick == 0 { + summary.dst.insert( + vpc(2), + PacketAndByte { + packets: 1_000, + bytes: 100_000, + }, + ); + } + let mut by_src = hashbrown::HashMap::new(); + by_src.insert(vpc(1), summary); + window.push(by_src); + } + let by_src: hashbrown::HashMap>> = + (&window).into(); + let smoothed = by_src + .get(&vpc(1)) + .unwrap_or_else(|| unreachable!()) + .smooth() + .unwrap_or_else(|_| unreachable!()); + let rate = smoothed + .dst + .get(&vpc(2)) + .unwrap_or_else(|| unreachable!()) + .packets; + assert!( + rate < 1.0, + "a destination silent for four of five ticks still read {rate} pkt/s" + ); + } + + #[test] + fn the_window_reads_back_in_push_order() { + bolero::check!() + .with_type() + .cloned() + .for_each(|pushes: Vec| { + let mut filter = SavitzkyGolayFilter::new(Duration::from_secs(1)); + for value in &pushes { + filter.push(*value); + } + let expect: Vec<_> = pushes.iter().rev().take(WINDOW).rev().copied().collect(); + let read: Vec<_> = filter.chronological().copied().collect(); + assert_eq!(read, expect); + }); + } + + #[test] + fn every_destination_holds_the_whole_window() { + bolero::check!().with_type().for_each( + |window: &SavitzkyGolayFilter>| match TransmitSummary::< + SavitzkyGolayFilter, + >::try_from( + window + ) { + Ok(converted) => { + for (dst, filter) in converted.dst.iter() { + assert_eq!( + filter.packets.data.len(), + WINDOW, + "destination {dst} holds a partial window" + ); + assert_eq!(filter.bytes.data.len(), WINDOW); + } + } + Err(DerivativeError::NotEnoughSamples(seen)) => assert!(seen < WINDOW), + Err(e) => panic!("{e}"), + }, + ); + } +} diff --git a/stats/src/scrape.rs b/stats/src/scrape.rs new file mode 100644 index 0000000000..cf9cdaef96 --- /dev/null +++ b/stats/src/scrape.rs @@ -0,0 +1,155 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright Open Network Fabric Authors + +use concurrency::sync::Arc; +use concurrency::sync::atomic::{AtomicUsize, Ordering}; +use concurrency::sync::{Mutex, MutexGuard}; +use std::collections::{BTreeMap, BTreeSet}; + +pub(crate) type Labels = BTreeMap; + +pub(crate) type SeriesId = (String, Labels); + +pub(crate) type Series = BTreeMap; + +pub(crate) type Shape = (String, Vec); + +#[derive(Debug, Default, Clone)] +pub(crate) struct Scrape { + held: Arc>, + shapes: Arc>>, + registrations: Arc, +} + +impl Scrape { + pub(crate) fn get(&self, name: &str, labels: &[(&str, &str)]) -> Option { + let labels: Labels = labels + .iter() + .map(|(k, v)| ((*k).to_string(), (*v).to_string())) + .collect(); + self.held().get(&(name.to_string(), labels)).copied() + } + + pub(crate) fn label_shapes(&self, name: &str) -> BTreeSet> { + self.shapes + .lock() + .iter() + .filter(|(series, _)| series == name) + .map(|(_, keys)| keys.clone()) + .collect() + } + + pub(crate) fn registrations(&self) -> usize { + self.registrations.load(Ordering::Relaxed) + } + + pub(crate) fn series(&self, name: &str) -> Vec<(Labels, f64)> { + self.held() + .iter() + .filter(|((series, _), _)| series == name) + .map(|((_, labels), value)| (labels.clone(), *value)) + .collect() + } + + pub(crate) fn nonzero(&self) -> Vec { + let mut out: Vec = self + .held() + .iter() + .filter(|&(_, &v)| v != 0.0) + .map(|((name, labels), value)| { + let labels: Vec = + labels.iter().map(|(k, v)| format!("{k}=\"{v}\"")).collect(); + format!("{name}{{{}}} = {value}", labels.join(",")) + }) + .collect(); + out.sort(); + out + } + + fn held(&self) -> MutexGuard<'_, Series> { + self.held.lock() + } +} + +#[derive(Debug)] +struct Cell { + at: SeriesId, + into: Arc>, +} + +impl Cell { + fn with(&self, f: impl FnOnce(&mut f64)) { + let mut held = self.into.lock(); + f(held.entry(self.at.clone()).or_insert(0.0)); + } +} + +impl metrics::GaugeFn for Cell { + fn increment(&self, value: f64) { + self.with(|held| *held += value); + } + + fn decrement(&self, value: f64) { + self.with(|held| *held -= value); + } + + fn set(&self, value: f64) { + self.with(|held| *held = value); + } +} + +impl metrics::Recorder for Scrape { + fn describe_counter( + &self, + _: metrics::KeyName, + _: Option, + _: metrics::SharedString, + ) { + } + + fn describe_gauge( + &self, + _: metrics::KeyName, + _: Option, + _: metrics::SharedString, + ) { + } + + fn describe_histogram( + &self, + _: metrics::KeyName, + _: Option, + _: metrics::SharedString, + ) { + } + + fn register_counter(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter { + metrics::Counter::noop() + } + + fn register_gauge(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge { + let labels = key + .labels() + .map(|label| (label.key().to_string(), label.value().to_string())) + .collect(); + let at = (key.name().to_string(), labels); + self.registrations.fetch_add(1, Ordering::Relaxed); + self.shapes.lock().insert(( + key.name().to_string(), + key.labels().map(|label| label.key().to_string()).collect(), + )); + self.held().entry(at.clone()).or_insert(0.0); + metrics::Gauge::from_arc(Arc::new(Cell { + at, + into: self.held.clone(), + })) + } + + fn register_histogram( + &self, + _: &metrics::Key, + _: &metrics::Metadata<'_>, + ) -> metrics::Histogram { + metrics::Histogram::noop() + } +} diff --git a/stats/src/vpc_stats.rs b/stats/src/vpc_stats.rs index c0e55d0552..b427b42ebd 100644 --- a/stats/src/vpc_stats.rs +++ b/stats/src/vpc_stats.rs @@ -35,6 +35,13 @@ pub struct FlowStats { pub drops: Counters, // drops (packets + optional bytes) } +#[derive(Debug, Default)] +pub struct StatsSnapshot { + pub names: HashMap, + pub pairs: Vec<(VpcPairKey, FlowStats)>, + pub vpcs: Vec<(VpcId, FlowStats)>, +} + #[derive(Debug, Default)] pub struct VpcStatsStore { /// Directional (src -> dst) @@ -143,6 +150,22 @@ impl VpcStatsStore { e.rate.bps = bps; } + pub async fn hand_over(&self, handovers: &[(VpcId, String)]) { + let forget: HashSet = handovers.iter().map(|(id, _)| *id).collect(); + { + let mut pairs = self.pair_stats.write().await; + pairs.retain(|(src, dst), _| !forget.contains(src) && !forget.contains(dst)); + } + { + let mut vpcs = self.vpc_stats.write().await; + vpcs.retain(|vpc, _| !forget.contains(vpc)); + } + let mut names = self.vpc_names.write(); + for (id, name) in handovers { + names.insert(*id, name.clone()); + } + } + pub async fn prune_to_vpcs(&self, alive: &HashSet) { { let mut pairs = self.pair_stats.write().await; @@ -158,6 +181,17 @@ impl VpcStatsStore { } } + pub async fn snapshot(&self) -> StatsSnapshot { + let pairs = self.pair_stats.read().await; + let vpcs = self.vpc_stats.read().await; + let names = self.vpc_names.read(); + StatsSnapshot { + pairs: pairs.iter().map(|(k, v)| (*k, *v)).collect(), + vpcs: vpcs.iter().map(|(k, v)| (*k, *v)).collect(), + names: names.clone(), + } + } + // ---------- Snapshots ---------- pub async fn snapshot_pairs(&self) -> Vec<(VpcPairKey, FlowStats)> { let map = self.pair_stats.read().await; @@ -175,3 +209,115 @@ impl VpcStatsStore { self.vpc_names.read().clone() } } + +#[cfg(test)] +mod under_readers { + use super::{VpcId, VpcStatsStore}; + use concurrency::sync::Arc; + use concurrency::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + use net::vxlan::Vni; + use vpcmap::VpcDiscriminant; + + const HANDOVERS: u64 = 20_000; + const SENT: u64 = 1_000; + + fn vpc(vni: u32) -> VpcId { + VpcDiscriminant::from_vni(Vni::new_checked(vni).unwrap_or_else(|_| unreachable!())) + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn a_reader_never_sees_a_name_against_the_previous_tenants_traffic() { + fn sent_by(tenant: u64) -> u64 { + SENT + tenant + } + + let store = VpcStatsStore::new(); + let (src, dst) = (vpc(100), vpc(200)); + let done = Arc::new(AtomicBool::new(false)); + let saw_the_window = Arc::new(AtomicU64::new(0)); + let saw_a_tenant_settled = Arc::new(AtomicU64::new(0)); + + let reader = tokio::spawn({ + let store = Arc::clone(&store); + let done = Arc::clone(&done); + let saw_the_window = Arc::clone(&saw_the_window); + let saw_a_tenant_settled = Arc::clone(&saw_a_tenant_settled); + async move { + while !done.load(Ordering::Relaxed) { + let taken = store.snapshot().await; + let (names, pairs) = (taken.names, taken.pairs); + + let Some(name) = names.get(&dst) else { + continue; + }; + let tenant: u64 = name.parse().unwrap_or_else(|e| unreachable!("{e:?}")); + let credited = pairs + .iter() + .find(|&&((from, to), _)| from == src && to == dst) + .map_or(0, |&(_, stats)| stats.ctr.packets); + + if credited == 0 { + saw_the_window.fetch_add(1, Ordering::Relaxed); + } else if credited == sent_by(tenant) { + saw_a_tenant_settled.fetch_add(1, Ordering::Relaxed); + } else { + panic!( + "tenant {tenant} sends {}, but is exported carrying {credited} -- \ + which is what tenant {} sent", + sent_by(tenant), + credited.saturating_sub(SENT) + ); + } + } + } + }); + + for tenant in 0..HANDOVERS { + store.hand_over(&[(dst, tenant.to_string())]).await; + tokio::task::yield_now().await; + store + .add_pair_counts(src, dst, sent_by(tenant), sent_by(tenant) * 500) + .await; + tokio::task::yield_now().await; + } + done.store(true, Ordering::Relaxed); + reader.await.unwrap_or_else(|e| unreachable!("{e:?}")); + + assert!( + saw_the_window.load(Ordering::Relaxed) > 0, + "no read landed between a handover and the incoming tenant's first packets" + ); + assert!( + saw_a_tenant_settled.load(Ordering::Relaxed) > 0, + "no read ever saw a tenant against its own traffic" + ); + } + + #[tokio::test] + async fn a_handover_drops_what_the_outgoing_tenant_accumulated() { + let store = VpcStatsStore::new(); + let (src, dst) = (vpc(100), vpc(200)); + store.add_pair_counts(src, dst, SENT, SENT * 500).await; + store.add_vpc_counts(dst, SENT, SENT * 500).await; + assert_eq!( + store + .snapshot_pairs() + .await + .first() + .map(|&(_, fs)| (fs.ctr.packets, fs.ctr.bytes)), + Some((SENT, SENT * 500)) + ); + + store.hand_over(&[(dst, "next".to_string())]).await; + + assert!( + store.snapshot_pairs().await.is_empty(), + "the outgoing tenant's pair counters survived the handover" + ); + assert!( + store.snapshot_vpcs().await.is_empty(), + "the outgoing tenant's totals survived the handover" + ); + assert_eq!(store.name_of(dst).as_deref(), Some("next")); + } +}