libp2p-scatter 0.3.0

broadcast messages to connected peers
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

use std::collections::HashMap;

use prometheus_client::metrics::counter::Counter;
use prometheus_client::metrics::family::Family;
use prometheus_client::metrics::gauge::Gauge;
use prometheus_client::registry::Registry;

use crate::Topic;

pub struct Metrics {
    /// Information needed to decide if a topic is allowed or not.
    topic_info: HashMap<Topic, EverSubscribed>,
    /// Status of our subscription to this topic. This metric allows analyzing other topic metrics
    /// filtered by our current subscription status.
    topic_subscription_status: Family<Topic, Gauge>,
    /// Number of peers subscribed to each topic. This allows us to analyze a topic's behaviour
    /// regardless of our subscription status.
    topic_peers_count: Family<Topic, Gauge>,

    /// Number of messages sent to each topic.
    topic_msg_sent_counts: Family<Topic, Counter>,
    /// Bytes from messages sent to each topic.
    topic_msg_sent_bytes: Family<Topic, Counter>,
    /// Number of messages published to each topic.
    topic_msg_published: Family<Topic, Counter>,

    /// Number of messages received on each topic
    topic_msg_recv_counts: Family<Topic, Counter>,
    /// Bytes received from messages for each topic.
    topic_msg_recv_bytes: Family<Topic, Counter>,
}

type EverSubscribed = bool;

impl Metrics {
    pub fn new(registry: &mut Registry) -> Self {
        macro_rules! register_family {
            ($name:expr, $help:expr) => {{
                let fam = Family::default();
                registry.register($name, $help, fam.clone());
                fam
            }};
        }

        let topic_subscription_status = register_family!(
            "topic_subscription_status",
            "Subscription status per known topic"
        );

        let topic_peers_count = register_family!(
            "topic_peers_counts",
            "Number of peers subscribed to each topic"
        );

        let topic_msg_sent_counts = register_family!(
            "topic_msg_sent_counts",
            "Number of gossip messages sent to each topic"
        );
        let topic_msg_published = register_family!(
            "topic_msg_published",
            "Number of gossip messages published to each topic"
        );
        let topic_msg_sent_bytes = register_family!(
            "topic_msg_sent_bytes",
            "Bytes from gossip messages sent to each topic"
        );

        let topic_msg_recv_counts = register_family!(
            "topic_msg_recv_counts",
            "Number of gossip messages received on each topic"
        );
        let topic_msg_recv_bytes = register_family!(
            "topic_msg_recv_bytes",
            "Bytes received from gossip messages for each topic"
        );

        Self {
            topic_info: HashMap::new(),
            topic_subscription_status,
            topic_peers_count,
            topic_msg_sent_counts,
            topic_msg_published,
            topic_msg_sent_bytes,
            topic_msg_recv_counts,
            topic_msg_recv_bytes,
        }
    }

    /// Registers a topic if not already known
    fn register_topic(&mut self, topic: &Topic) {
        if !self.topic_info.contains_key(topic) {
            self.topic_info.entry(*topic).or_insert(false);
            self.topic_subscription_status.get_or_create(topic).set(0);
        }
    }

    pub(crate) fn subscribe(&mut self, topic: &Topic) {
        self.register_topic(topic);
        self.topic_info.entry(*topic).or_insert(false);
        self.topic_subscription_status.get_or_create(topic).set(1);
    }

    pub(crate) fn unsubscribe(&mut self, topic: &Topic) {
        self.register_topic(topic);
        self.topic_subscription_status.get_or_create(topic).set(0);
    }

    /// Increase the number of peers that are subscribed to this topic.
    pub(crate) fn inc_topic_peers(&mut self, topic: &Topic) {
        self.register_topic(topic);
        self.topic_peers_count.get_or_create(topic).inc();
    }

    /// Decrease the number of peers that are subscribed to this topic.
    pub(crate) fn dec_topic_peers(&mut self, topic: &Topic) {
        self.register_topic(topic);
        self.topic_peers_count.get_or_create(topic).dec();
    }

    pub(crate) fn register_published_message(&mut self, topic: &Topic) {
        self.register_topic(topic);
        self.topic_msg_published.get_or_create(topic).inc();
    }

    /// Register sending a message over a topic.
    pub(crate) fn msg_sent(&mut self, topic: &Topic, bytes: usize) {
        self.register_topic(topic);
        self.topic_msg_sent_counts.get_or_create(topic).inc();
        self.topic_msg_sent_bytes
            .get_or_create(topic)
            .inc_by(bytes as u64);
    }

    /// Register that a message was received .
    pub(crate) fn msg_received(&mut self, topic: &Topic, bytes: usize) {
        self.register_topic(topic);
        self.topic_msg_recv_counts.get_or_create(topic).inc();
        self.topic_msg_recv_bytes
            .get_or_create(topic)
            .inc_by(bytes as u64);
    }
}