use crate::Event;
use futures::{prelude::*, channel::mpsc, ready, stream::FusedStream};
use parking_lot::Mutex;
use prometheus_endpoint::{register, CounterVec, GaugeVec, Opts, PrometheusError, Registry, U64};
use std::{
convert::TryFrom as _,
fmt, pin::Pin, sync::Arc,
task::{Context, Poll}
};
pub fn channel(name: &'static str) -> (Sender, Receiver) {
let (tx, rx) = mpsc::unbounded();
let metrics = Arc::new(Mutex::new(None));
let tx = Sender { inner: tx, name, metrics: metrics.clone() };
let rx = Receiver { inner: rx, name, metrics };
(tx, rx)
}
pub struct Sender {
inner: mpsc::UnboundedSender<Event>,
name: &'static str,
metrics: Arc<Mutex<Option<Arc<Option<Metrics>>>>>,
}
impl fmt::Debug for Sender {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_tuple("Sender").finish()
}
}
impl Drop for Sender {
fn drop(&mut self) {
let metrics = self.metrics.lock();
if let Some(Some(metrics)) = metrics.as_ref().map(|m| &**m) {
metrics.num_channels.with_label_values(&[self.name]).dec();
}
}
}
pub struct Receiver {
inner: mpsc::UnboundedReceiver<Event>,
name: &'static str,
metrics: Arc<Mutex<Option<Arc<Option<Metrics>>>>>,
}
impl Stream for Receiver {
type Item = Event;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Event>> {
if let Some(ev) = ready!(Pin::new(&mut self.inner).poll_next(cx)) {
let metrics = self.metrics.lock().clone();
match metrics.as_ref().map(|m| m.as_ref()) {
Some(Some(metrics)) => metrics.event_out(&ev, self.name),
Some(None) => (),
None => log::warn!("Inconsistency in out_events: event happened before sender associated"),
}
Poll::Ready(Some(ev))
} else {
Poll::Ready(None)
}
}
}
impl fmt::Debug for Receiver {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_tuple("Receiver").finish()
}
}
impl Drop for Receiver {
fn drop(&mut self) {
if !self.inner.is_terminated() {
while let Some(Some(_)) = self.next().now_or_never() {}
}
}
}
pub struct OutChannels {
event_streams: Vec<Sender>,
metrics: Arc<Option<Metrics>>,
}
impl OutChannels {
pub fn new(registry: Option<&Registry>) -> Result<Self, PrometheusError> {
let metrics = if let Some(registry) = registry {
Some(Metrics::register(registry)?)
} else {
None
};
Ok(OutChannels {
event_streams: Vec::new(),
metrics: Arc::new(metrics),
})
}
pub fn push(&mut self, sender: Sender) {
let mut metrics = sender.metrics.lock();
debug_assert!(metrics.is_none());
*metrics = Some(self.metrics.clone());
drop(metrics);
if let Some(metrics) = &*self.metrics {
metrics.num_channels.with_label_values(&[sender.name]).inc();
}
self.event_streams.push(sender);
}
pub fn send(&mut self, event: Event) {
self.event_streams.retain(|sender| {
sender.inner.unbounded_send(event.clone()).is_ok()
});
if let Some(metrics) = &*self.metrics {
for ev in &self.event_streams {
metrics.event_in(&event, 1, ev.name);
}
}
}
}
impl fmt::Debug for OutChannels {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.debug_struct("OutChannels")
.field("num_channels", &self.event_streams.len())
.finish()
}
}
struct Metrics {
events_total: CounterVec<U64>,
notifications_sizes: CounterVec<U64>,
num_channels: GaugeVec<U64>,
}
impl Metrics {
fn register(registry: &Registry) -> Result<Self, PrometheusError> {
Ok(Self {
events_total: register(CounterVec::new(
Opts::new(
"sub_libp2p_out_events_events_total",
"Number of broadcast network events that have been sent or received across all \
channels"
),
&["event_name", "action", "name"]
)?, registry)?,
notifications_sizes: register(CounterVec::new(
Opts::new(
"sub_libp2p_out_events_notifications_sizes",
"Size of notification events that have been sent or received across all \
channels"
),
&["protocol", "action", "name"]
)?, registry)?,
num_channels: register(GaugeVec::new(
Opts::new(
"sub_libp2p_out_events_num_channels",
"Number of internal active channels that broadcast network events",
),
&["name"]
)?, registry)?,
})
}
fn event_in(&self, event: &Event, num: u64, name: &str) {
match event {
Event::Dht(_) => {
self.events_total
.with_label_values(&["dht", "sent", name])
.inc_by(num);
}
Event::SyncConnected { .. } => {
self.events_total
.with_label_values(&["sync-connected", "sent", name])
.inc_by(num);
}
Event::SyncDisconnected { .. } => {
self.events_total
.with_label_values(&["sync-disconnected", "sent", name])
.inc_by(num);
}
Event::NotificationStreamOpened { protocol, .. } => {
self.events_total
.with_label_values(&[&format!("notif-open-{:?}", protocol), "sent", name])
.inc_by(num);
},
Event::NotificationStreamClosed { protocol, .. } => {
self.events_total
.with_label_values(&[&format!("notif-closed-{:?}", protocol), "sent", name])
.inc_by(num);
},
Event::NotificationsReceived { messages, .. } => {
for (protocol, message) in messages {
self.events_total
.with_label_values(&[&format!("notif-{:?}", protocol), "sent", name])
.inc_by(num);
self.notifications_sizes
.with_label_values(&[protocol, "sent", name])
.inc_by(num.saturating_mul(u64::try_from(message.len()).unwrap_or(u64::max_value())));
}
},
}
}
fn event_out(&self, event: &Event, name: &str) {
match event {
Event::Dht(_) => {
self.events_total
.with_label_values(&["dht", "received", name])
.inc();
}
Event::SyncConnected { .. } => {
self.events_total
.with_label_values(&["sync-connected", "received", name])
.inc();
}
Event::SyncDisconnected { .. } => {
self.events_total
.with_label_values(&["sync-disconnected", "received", name])
.inc();
}
Event::NotificationStreamOpened { protocol, .. } => {
self.events_total
.with_label_values(&[&format!("notif-open-{:?}", protocol), "received", name])
.inc();
},
Event::NotificationStreamClosed { protocol, .. } => {
self.events_total
.with_label_values(&[&format!("notif-closed-{:?}", protocol), "received", name])
.inc();
},
Event::NotificationsReceived { messages, .. } => {
for (protocol, message) in messages {
self.events_total
.with_label_values(&[&format!("notif-{:?}", protocol), "received", name])
.inc();
self.notifications_sizes
.with_label_values(&[&protocol, "received", name])
.inc_by(u64::try_from(message.len()).unwrap_or(u64::max_value()));
}
},
}
}
}