-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: export quinn connection metrics (#23)
* log ts for debug secure handshake * feat: add metric collector service * chore: remove manual count stream * chore: resolve comment for PR * chore: change log stats to debug level * feat: add sent_pkt and largest_mtu stats to metrics * chore: export peerConnectionMetrics, that can use by other service
- Loading branch information
1 parent
5e1d3b4
commit 57df7cd
Showing
8 changed files
with
192 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| use std::{collections::VecDeque, time::Duration}; | ||
|
|
||
| use serde::{Deserialize, Serialize}; | ||
| use tokio::{select, time::Interval}; | ||
|
|
||
| use crate::{peer::PeerConnectionMetric, ConnectionId, ErrorExt, P2pServiceEvent, PeerId}; | ||
|
|
||
| use super::P2pService; | ||
|
|
||
| #[derive(Debug, PartialEq, Eq)] | ||
| pub enum MetricsServiceEvent { | ||
| OnPeerConnectionMetric(PeerId, Vec<(ConnectionId, PeerId, PeerConnectionMetric)>), | ||
| } | ||
|
|
||
| #[derive(Deserialize, Serialize)] | ||
| enum Message { | ||
| Scan, | ||
| Info(Vec<(ConnectionId, PeerId, PeerConnectionMetric)>), | ||
| } | ||
|
|
||
| const DEFAULT_COLLECTOR_INTERVAL: u64 = 1; | ||
|
|
||
| pub struct MetricsService { | ||
| is_collector: bool, | ||
| service: P2pService, | ||
| ticker: Interval, | ||
| outs: VecDeque<MetricsServiceEvent>, | ||
| } | ||
|
|
||
| impl MetricsService { | ||
| pub fn new(collect_interval: Option<Duration>, service: P2pService, is_collector: bool) -> Self { | ||
| let ticker = tokio::time::interval(collect_interval.unwrap_or(Duration::from_secs(DEFAULT_COLLECTOR_INTERVAL))); | ||
|
|
||
| Self { | ||
| is_collector, | ||
| ticker, | ||
| service, | ||
| outs: VecDeque::new(), | ||
| } | ||
| } | ||
|
|
||
| pub async fn recv(&mut self) -> anyhow::Result<MetricsServiceEvent> { | ||
| loop { | ||
| if let Some(out) = self.outs.pop_front() { | ||
| return Ok(out); | ||
| } | ||
|
|
||
| select! { | ||
| _ = self.ticker.tick() => { | ||
| if self.is_collector { | ||
| let metrics = self.service.ctx.metrics(); | ||
| self.outs.push_back(MetricsServiceEvent::OnPeerConnectionMetric(self.service.router().local_id(), metrics)); | ||
|
|
||
| let requester = self.service.requester(); | ||
| tokio::spawn(async move { | ||
| requester.send_broadcast(bincode::serialize(&Message::Scan).expect("should convert to buf")).await; | ||
| }); | ||
| } | ||
| } | ||
| event = self.service.recv() => match event.expect("should work") { | ||
| P2pServiceEvent::Unicast(from, data) | P2pServiceEvent::Broadcast(from, data) => { | ||
| if let Ok(msg) = bincode::deserialize::<Message>(&data) { | ||
| match msg { | ||
| Message::Scan => { | ||
| let metrics = self.service.ctx.metrics(); | ||
| let requester = self.service.requester(); | ||
| tokio::spawn(async move { | ||
| requester.try_send_unicast(from, bincode::serialize(&Message::Info(metrics)).expect("should convert to buf")) | ||
| .await | ||
| .print_on_err("send metrics info to collector error"); | ||
| }); | ||
| } | ||
| Message::Info(peer_metrics) => { | ||
| self.outs.push_back(MetricsServiceEvent::OnPeerConnectionMetric(from, peer_metrics)); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| P2pServiceEvent::Stream(..) => {} | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| use std::time::Duration; | ||
|
|
||
| use crate::{ | ||
| metrics_service::{MetricsService, MetricsServiceEvent}, | ||
| PeerId, | ||
| }; | ||
| use test_log::test; | ||
|
|
||
| use super::create_node; | ||
|
|
||
| #[test(tokio::test)] | ||
| async fn metric_collect() { | ||
| let (mut node1, addr1) = create_node(true, 1, vec![]).await; | ||
| let mut service1 = MetricsService::new(None, node1.create_service(0.into()), true); | ||
| tokio::spawn(async move { while node1.recv().await.is_ok() {} }); | ||
|
|
||
| let (mut node2, _) = create_node(true, 2, vec![addr1.clone()]).await; | ||
| let mut service2 = MetricsService::new(None, node2.create_service(0.into()), false); | ||
| tokio::spawn(async move { while node2.recv().await.is_ok() {} }); | ||
| tokio::spawn(async move { while service2.recv().await.is_ok() {} }); | ||
|
|
||
| tokio::time::sleep(Duration::from_secs(1)).await; | ||
|
|
||
| let events = vec![ | ||
| tokio::time::timeout(Duration::from_secs(3), service1.recv()).await.expect("").expect(""), | ||
| tokio::time::timeout(Duration::from_secs(3), service1.recv()).await.expect("").expect(""), | ||
| tokio::time::timeout(Duration::from_secs(3), service1.recv()).await.expect("").expect(""), | ||
| tokio::time::timeout(Duration::from_secs(3), service1.recv()).await.expect("").expect(""), | ||
| ]; | ||
|
|
||
| let event_from_peers: Vec<(PeerId, usize)> = events | ||
| .iter() | ||
| .map(|e| match e { | ||
| MetricsServiceEvent::OnPeerConnectionMetric(peer, metrics) => (*peer, metrics.len()), | ||
| }) | ||
| .collect(); | ||
|
|
||
| println!("{:?}", events); | ||
|
|
||
| assert_eq!(event_from_peers, vec![(PeerId(1), 1), (PeerId(1), 1), (PeerId(2), 1), (PeerId(2), 1)]); | ||
| } |