use std::{
collections::HashSet,
sync::{
atomic::{AtomicBool, Ordering},
Mutex,
},
};
use nonempty_collections::NEVec;
use zenoh_config::{DataMessage, DownsamplingItemConf, DownsamplingRuleConf};
use zenoh_core::zlock;
use zenoh_keyexpr::keyexpr_tree::{
impls::KeyedSetProvider, support::UnknownWildness, IKeyExprTree, IKeyExprTreeMut, KeBoxTree,
};
use zenoh_protocol::{
network::{NetworkBodyMut, Push},
zenoh::PushBody,
};
use zenoh_result::ZResult;
use crate::net::routing::interceptor::*;
pub(crate) fn downsampling_interceptor_factories(
config: &Vec<DownsamplingItemConf>,
) -> ZResult<Vec<InterceptorFactory>> {
let mut res: Vec<InterceptorFactory> = vec![];
let mut id_set = HashSet::new();
for ds in config {
if let Some(id) = &ds.id {
if !id_set.insert(id.clone()) {
bail!("Invalid Downsampling config: id '{id}' is repeated");
}
}
res.push(Box::new(DownsamplingInterceptorFactory::new(ds.clone())));
}
Ok(res)
}
pub struct DownsamplingInterceptorFactory {
interfaces: Option<NEVec<String>>,
link_protocols: Option<NEVec<InterceptorLink>>,
rules: NEVec<DownsamplingRuleConf>,
flows: InterfaceEnabled,
messages: DownsamplingFilters,
}
impl DownsamplingInterceptorFactory {
pub fn new(conf: DownsamplingItemConf) -> Self {
Self {
interfaces: conf.interfaces,
rules: conf.rules,
link_protocols: conf.link_protocols,
flows: conf.flows.map(|f| (&f).into()).unwrap_or(InterfaceEnabled {
ingress: true,
egress: true,
}),
messages: DownsamplingFilters::new(conf.messages.iter().copied()),
}
}
}
impl InterceptorFactoryTrait for DownsamplingInterceptorFactory {
fn new_transport_unicast(
&self,
transport: &TransportUnicast,
) -> (Option<IngressInterceptor>, Option<EgressInterceptor>) {
if let Some(interfaces) = &self.interfaces {
if let Ok(links) = transport.get_links() {
for link in links {
if !link.interfaces.iter().any(|x| interfaces.contains(x)) {
return (None, None);
}
}
}
}
if let Some(config_protocols) = &self.link_protocols {
match transport.get_auth_ids() {
Ok(auth_ids) => {
if !auth_ids
.link_auth_ids()
.iter()
.map(|auth_id| InterceptorLinkWrapper::from(auth_id).0)
.any(|v| config_protocols.contains(&v))
{
return (None, None);
}
}
Err(e) => {
tracing::error!("Error loading transport AuthIds: {e}");
return (None, None);
}
}
}
#[cfg(feature = "stats")]
let Ok(stats) = transport
.get_stats()
.map(|stats| stats.drop_stats(zenoh_stats::ReasonLabel::Downsampling))
else {
return (None, None);
};
let interceptor = |flow| {
let direction = match flow {
InterceptorFlow::Ingress => "ingress",
InterceptorFlow::Egress => "egress",
};
tracing::debug!("New {direction} downsampler on transport unicast {transport:?}");
Box::new(DownsamplingInterceptor::new(
self.messages,
&self.rules,
flow,
#[cfg(feature = "stats")]
stats.clone(),
)) as Interceptor
};
(
self.flows
.ingress
.then(|| interceptor(InterceptorFlow::Ingress)),
self.flows
.egress
.then(|| interceptor(InterceptorFlow::Egress)),
)
}
fn new_transport_multicast(
&self,
_transport: &TransportMulticast,
) -> Option<EgressInterceptor> {
None
}
fn new_peer_multicast(&self, _transport: &TransportMulticast) -> Option<IngressInterceptor> {
None
}
}
#[derive(Debug, Default, Clone, Copy)]
pub(crate) struct DownsamplingFilters {
delete: bool,
put: bool,
query: bool,
reply: bool,
}
impl DownsamplingFilters {
fn new<T: IntoIterator<Item = DataMessage>>(iter: T) -> Self {
let mut filters = Self::default();
for m in iter {
match m {
DataMessage::Delete => filters.delete = true,
DataMessage::Put => filters.put = true,
DataMessage::Query => filters.query = true,
DataMessage::Reply => filters.reply = true,
}
}
filters
}
fn is_msg_filtered(&self, msg: &mut NetworkMessageMut) -> bool {
match msg.body {
NetworkBodyMut::Push(Push { payload, .. }) => match payload {
PushBody::Put(_) => self.put,
PushBody::Del(_) => self.delete,
},
NetworkBodyMut::Request(_) => self.query,
NetworkBodyMut::Response(_) => self.reply,
NetworkBodyMut::ResponseFinal(_) => false,
NetworkBodyMut::Interest(_) => false,
NetworkBodyMut::Declare(_) => false,
NetworkBodyMut::OAM(_) => false,
}
}
}
struct TimeState {
pub threshold: std::time::Duration,
pub latest_message_timestamp: Mutex<std::time::Instant>,
}
pub(crate) struct DownsamplingInterceptor {
filters: DownsamplingFilters,
ke_id: KeBoxTree<usize, UnknownWildness, KeyedSetProvider>,
ke_state: Vec<TimeState>,
flow: InterceptorFlow,
#[cfg(feature = "stats")]
stats: zenoh_stats::DropStats,
}
impl DownsamplingInterceptor {
fn get_id(&self, key_expr: &keyexpr) -> Option<usize> {
let node = self.ke_id.intersecting_keys(key_expr).next()?;
self.ke_id.weight_at(&node).copied()
}
}
static INFO_FLAG: AtomicBool = AtomicBool::new(false);
impl InterceptorTrait for DownsamplingInterceptor {
fn compute_keyexpr_cache(&self, key_expr: &keyexpr) -> Option<Box<dyn Any + Send + Sync>> {
Some(Box::new(self.get_id(key_expr)))
}
fn intercept(&self, msg: &mut NetworkMessageMut, ctx: &mut dyn InterceptorContext) -> bool {
if !self.filters.is_msg_filtered(msg) {
return true;
};
let cache = ctx
.get_cache(msg)
.and_then(|c| c.downcast_ref::<Option<usize>>().copied());
let Some(id) = cache.unwrap_or_else(|| self.get_id(&ctx.full_keyexpr(msg)?)) else {
return true;
};
let Some(state) = self.ke_state.get(id) else {
tracing::debug!("unexpected cache ID {id}");
return true;
};
let mut latest_message_timestamp = zlock!(state.latest_message_timestamp);
let timestamp = std::time::Instant::now();
if timestamp - *latest_message_timestamp >= state.threshold {
*latest_message_timestamp = timestamp;
true
} else {
drop(latest_message_timestamp);
if !INFO_FLAG.swap(true, Ordering::Relaxed) {
tracing::info!("Some message(s) have been dropped by the downsampling interceptor. Enable trace level tracing for more details.");
}
let direction = match self.flow {
InterceptorFlow::Egress => "to",
InterceptorFlow::Ingress => "from",
};
tracing::trace!(
"Message dropped by the downsampling interceptor: {msg}({key_expr}) {direction}:{face}",
key_expr = ctx.full_expr(msg).unwrap_or_default(),
face = ctx.face().map(|f| f.to_string()).unwrap_or_default(),
);
#[cfg(feature = "stats")]
self.stats
.observe_network_message_dropped_payload(stats_direction(self.flow), msg);
false
}
}
}
const NANOS_PER_SEC: f64 = 1_000_000_000.0;
impl DownsamplingInterceptor {
pub fn new(
filters: DownsamplingFilters,
rules: &NEVec<DownsamplingRuleConf>,
flow: InterceptorFlow,
#[cfg(feature = "stats")] stats: zenoh_stats::DropStats,
) -> Self {
let mut ke_id = KeBoxTree::default();
let mut ke_state = Vec::new();
for (id, rule) in rules.into_iter().enumerate() {
let mut threshold = std::time::Duration::MAX;
let mut latest_message_timestamp = std::time::Instant::now();
if rule.freq != 0.0 {
threshold =
std::time::Duration::from_nanos((1. / rule.freq * NANOS_PER_SEC) as u64);
latest_message_timestamp -= threshold;
}
ke_id.insert(&rule.key_expr, id);
ke_state.push(TimeState {
threshold,
latest_message_timestamp: latest_message_timestamp.into(),
});
tracing::debug!(
"New downsampler rule enabled: key_expr={key_expr:?}, threshold={threshold:?}, messages={filters:?}",
key_expr = rule.key_expr,
);
}
Self {
filters,
ke_id,
ke_state,
flow,
#[cfg(feature = "stats")]
stats,
}
}
}