mirror of
https://github.com/YGGverse/aquatic.git
synced 2026-03-31 09:45:31 +00:00
http: upgrade metrics crate to 0.22
This commit is contained in:
Joakim Frostegård
parent
fe6a7ef8b5
commit
c7f7f010ca
6 changed files with 90 additions and 161 deletions
|
|
@ -26,9 +26,9 @@ use once_cell::sync::Lazy;
|
|||
use crate::common::*;
|
||||
use crate::config::Config;
|
||||
|
||||
use super::request::{parse_request, RequestParseError};
|
||||
#[cfg(feature = "metrics")]
|
||||
use super::{peer_addr_to_ip_version_str, WORKER_INDEX};
|
||||
use super::peer_addr_to_ip_version_str;
|
||||
use super::request::{parse_request, RequestParseError};
|
||||
|
||||
const REQUEST_BUFFER_SIZE: usize = 2048;
|
||||
const RESPONSE_BUFFER_SIZE: usize = 4096;
|
||||
|
|
@ -67,6 +67,7 @@ pub enum ConnectionError {
|
|||
Other(#[from] anyhow::Error),
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub(super) async fn run_connection(
|
||||
config: Rc<Config>,
|
||||
access_list: Arc<AccessListArcSwap>,
|
||||
|
|
@ -75,6 +76,7 @@ pub(super) async fn run_connection(
|
|||
opt_tls_config: Option<Arc<ArcSwap<RustlsConfig>>>,
|
||||
valid_until: Rc<RefCell<ValidUntil>>,
|
||||
stream: TcpStream,
|
||||
worker_index: usize,
|
||||
) -> Result<(), ConnectionError> {
|
||||
let access_list_cache = create_access_list_cache(&access_list);
|
||||
let request_buffer = Box::new([0u8; REQUEST_BUFFER_SIZE]);
|
||||
|
|
@ -114,6 +116,7 @@ pub(super) async fn run_connection(
|
|||
request_buffer_position: 0,
|
||||
response_buffer,
|
||||
stream,
|
||||
worker_index_string: worker_index.to_string(),
|
||||
};
|
||||
|
||||
conn.run().await
|
||||
|
|
@ -130,6 +133,7 @@ pub(super) async fn run_connection(
|
|||
request_buffer_position: 0,
|
||||
response_buffer,
|
||||
stream,
|
||||
worker_index_string: worker_index.to_string(),
|
||||
};
|
||||
|
||||
conn.run().await
|
||||
|
|
@ -148,6 +152,7 @@ struct Connection<S> {
|
|||
request_buffer_position: usize,
|
||||
response_buffer: Box<[u8; RESPONSE_BUFFER_SIZE]>,
|
||||
stream: S,
|
||||
worker_index_string: String,
|
||||
}
|
||||
|
||||
impl<S> Connection<S>
|
||||
|
|
@ -244,12 +249,13 @@ where
|
|||
match request {
|
||||
Request::Announce(request) => {
|
||||
#[cfg(feature = "metrics")]
|
||||
::metrics::increment_counter!(
|
||||
::metrics::counter!(
|
||||
"aquatic_requests_total",
|
||||
"type" => "announce",
|
||||
"ip_version" => peer_addr_to_ip_version_str(&peer_addr),
|
||||
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
|
||||
);
|
||||
"worker_index" => self.worker_index_string.clone(),
|
||||
)
|
||||
.increment(1);
|
||||
|
||||
let info_hash = request.info_hash;
|
||||
|
||||
|
|
@ -291,12 +297,13 @@ where
|
|||
}
|
||||
Request::Scrape(ScrapeRequest { info_hashes }) => {
|
||||
#[cfg(feature = "metrics")]
|
||||
::metrics::increment_counter!(
|
||||
::metrics::counter!(
|
||||
"aquatic_requests_total",
|
||||
"type" => "scrape",
|
||||
"ip_version" => peer_addr_to_ip_version_str(&peer_addr),
|
||||
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
|
||||
);
|
||||
"worker_index" => self.worker_index_string.clone(),
|
||||
)
|
||||
.increment(1);
|
||||
|
||||
let mut info_hashes_by_worker: BTreeMap<usize, Vec<InfoHash>> = BTreeMap::new();
|
||||
|
||||
|
|
@ -438,12 +445,13 @@ where
|
|||
.opt_peer_addr
|
||||
.expect("peer addr should already have been extracted by now");
|
||||
|
||||
::metrics::increment_counter!(
|
||||
::metrics::counter!(
|
||||
"aquatic_responses_total",
|
||||
"type" => response_type,
|
||||
"ip_version" => peer_addr_to_ip_version_str(&peer_addr),
|
||||
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
|
||||
);
|
||||
"worker_index" => self.worker_index_string.clone(),
|
||||
)
|
||||
.increment(1);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
|
|
|
|||
|
|
@ -25,9 +25,6 @@ use crate::common::*;
|
|||
use crate::config::Config;
|
||||
use crate::workers::socket::connection::{run_connection, ConnectionError};
|
||||
|
||||
#[cfg(feature = "metrics")]
|
||||
thread_local! { static WORKER_INDEX: ::std::cell::Cell<usize> = Default::default() }
|
||||
|
||||
struct ConnectionHandle {
|
||||
close_conn_sender: LocalSender<()>,
|
||||
valid_until: Rc<RefCell<ValidUntil>>,
|
||||
|
|
@ -44,9 +41,6 @@ pub async fn run_socket_worker(
|
|||
server_start_instant: ServerStartInstant,
|
||||
worker_index: usize,
|
||||
) {
|
||||
#[cfg(feature = "metrics")]
|
||||
WORKER_INDEX.with(|index| index.set(worker_index));
|
||||
|
||||
let config = Rc::new(config);
|
||||
let access_list = state.access_list;
|
||||
|
||||
|
|
@ -93,12 +87,14 @@ pub async fn run_socket_worker(
|
|||
)
|
||||
async move {
|
||||
#[cfg(feature = "metrics")]
|
||||
::metrics::increment_gauge!(
|
||||
let active_connections_gauge = ::metrics::gauge!(
|
||||
"aquatic_active_connections",
|
||||
1.0,
|
||||
"worker_index" => worker_index.to_string(),
|
||||
);
|
||||
|
||||
#[cfg(feature = "metrics")]
|
||||
active_connections_gauge.increment(1.0);
|
||||
|
||||
let f1 = async { run_connection(
|
||||
config,
|
||||
access_list,
|
||||
|
|
@ -107,6 +103,7 @@ pub async fn run_socket_worker(
|
|||
opt_tls_config,
|
||||
valid_until.clone(),
|
||||
stream,
|
||||
worker_index,
|
||||
).await
|
||||
};
|
||||
let f2 = async {
|
||||
|
|
@ -118,11 +115,7 @@ pub async fn run_socket_worker(
|
|||
let result = race(f1, f2).await;
|
||||
|
||||
#[cfg(feature = "metrics")]
|
||||
::metrics::decrement_gauge!(
|
||||
"aquatic_active_connections",
|
||||
1.0,
|
||||
"worker_index" => worker_index.to_string(),
|
||||
);
|
||||
active_connections_gauge.decrement(1.0);
|
||||
|
||||
match result {
|
||||
Ok(()) => (),
|
||||
|
|
|
|||
|
|
@ -18,9 +18,6 @@ use crate::config::Config;
|
|||
|
||||
use self::storage::TorrentMaps;
|
||||
|
||||
#[cfg(feature = "metrics")]
|
||||
thread_local! { static WORKER_INDEX: ::std::cell::Cell<usize> = Default::default() }
|
||||
|
||||
pub async fn run_swarm_worker(
|
||||
_sentinel: PanicSentinel,
|
||||
config: Config,
|
||||
|
|
@ -29,12 +26,9 @@ pub async fn run_swarm_worker(
|
|||
server_start_instant: ServerStartInstant,
|
||||
worker_index: usize,
|
||||
) {
|
||||
#[cfg(feature = "metrics")]
|
||||
WORKER_INDEX.with(|index| index.set(worker_index));
|
||||
|
||||
let (_, mut request_receivers) = request_mesh_builder.join(Role::Consumer).await.unwrap();
|
||||
|
||||
let torrents = Rc::new(RefCell::new(TorrentMaps::default()));
|
||||
let torrents = Rc::new(RefCell::new(TorrentMaps::new(worker_index)));
|
||||
let access_list = state.access_list;
|
||||
|
||||
// Periodically clean torrents
|
||||
|
|
@ -69,16 +63,14 @@ pub async fn run_swarm_worker(
|
|||
|
||||
::metrics::gauge!(
|
||||
"aquatic_torrents",
|
||||
torrents.ipv4.len() as f64,
|
||||
"ip_version" => "4",
|
||||
"worker_index" => worker_index.to_string(),
|
||||
);
|
||||
).set(torrents.ipv4.len() as f64);
|
||||
::metrics::gauge!(
|
||||
"aquatic_torrents",
|
||||
torrents.ipv6.len() as f64,
|
||||
"ip_version" => "6",
|
||||
"worker_index" => worker_index.to_string(),
|
||||
);
|
||||
).set(torrents.ipv6.len() as f64);
|
||||
|
||||
Some(Duration::from_secs(config.metrics.torrent_count_update_interval))
|
||||
})()
|
||||
|
|
|
|||
|
|
@ -16,7 +16,6 @@ use aquatic_http_protocol::response::*;
|
|||
use crate::config::Config;
|
||||
|
||||
#[cfg(feature = "metrics")]
|
||||
use crate::workers::swarm::WORKER_INDEX;
|
||||
|
||||
pub trait Ip: ::std::fmt::Debug + Copy + Eq + ::std::hash::Hash {
|
||||
#[cfg(feature = "metrics")]
|
||||
|
|
@ -36,13 +35,35 @@ impl Ip for Ipv6Addr {
|
|||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct TorrentMaps {
|
||||
pub ipv4: TorrentMap<Ipv4Addr>,
|
||||
pub ipv6: TorrentMap<Ipv6Addr>,
|
||||
#[cfg(feature = "metrics")]
|
||||
pub ipv4_peer_gauge: metrics::Gauge,
|
||||
#[cfg(feature = "metrics")]
|
||||
pub ipv6_peer_gauge: metrics::Gauge,
|
||||
}
|
||||
|
||||
impl TorrentMaps {
|
||||
pub fn new(worker_index: usize) -> Self {
|
||||
Self {
|
||||
ipv4: Default::default(),
|
||||
ipv6: Default::default(),
|
||||
#[cfg(feature = "metrics")]
|
||||
ipv4_peer_gauge: ::metrics::gauge!(
|
||||
"aquatic_peers",
|
||||
"ip_version" => "4",
|
||||
"worker_index" => worker_index.to_string(),
|
||||
),
|
||||
#[cfg(feature = "metrics")]
|
||||
ipv6_peer_gauge: ::metrics::gauge!(
|
||||
"aquatic_peers",
|
||||
"ip_version" => "6",
|
||||
"worker_index" => worker_index.to_string(),
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn handle_announce_request(
|
||||
&mut self,
|
||||
config: &Config,
|
||||
|
|
@ -63,6 +84,8 @@ impl TorrentMaps {
|
|||
peer_ip_address,
|
||||
request,
|
||||
valid_until,
|
||||
#[cfg(feature = "metrics")]
|
||||
&self.ipv4_peer_gauge,
|
||||
);
|
||||
|
||||
AnnounceResponse {
|
||||
|
|
@ -85,6 +108,8 @@ impl TorrentMaps {
|
|||
peer_ip_address,
|
||||
request,
|
||||
valid_until,
|
||||
#[cfg(feature = "metrics")]
|
||||
&self.ipv6_peer_gauge,
|
||||
);
|
||||
|
||||
AnnounceResponse {
|
||||
|
|
@ -157,8 +182,20 @@ impl TorrentMaps {
|
|||
|
||||
let now = server_start_instant.seconds_elapsed();
|
||||
|
||||
Self::clean_torrent_map(config, &mut access_list_cache, &mut self.ipv4, now);
|
||||
Self::clean_torrent_map(config, &mut access_list_cache, &mut self.ipv6, now);
|
||||
Self::clean_torrent_map(
|
||||
config,
|
||||
&mut access_list_cache,
|
||||
&mut self.ipv4,
|
||||
now,
|
||||
&self.ipv4_peer_gauge,
|
||||
);
|
||||
Self::clean_torrent_map(
|
||||
config,
|
||||
&mut access_list_cache,
|
||||
&mut self.ipv6,
|
||||
now,
|
||||
&self.ipv6_peer_gauge,
|
||||
);
|
||||
}
|
||||
|
||||
fn clean_torrent_map<I: Ip>(
|
||||
|
|
@ -166,6 +203,7 @@ impl TorrentMaps {
|
|||
access_list_cache: &mut AccessListCache,
|
||||
torrent_map: &mut TorrentMap<I>,
|
||||
now: SecondsSinceServerStart,
|
||||
#[cfg(feature = "metrics")] peer_gauge: &::metrics::Gauge,
|
||||
) {
|
||||
let mut total_num_peers = 0;
|
||||
|
||||
|
|
@ -197,12 +235,7 @@ impl TorrentMaps {
|
|||
let total_num_peers = total_num_peers as f64;
|
||||
|
||||
#[cfg(feature = "metrics")]
|
||||
::metrics::gauge!(
|
||||
"aquatic_peers",
|
||||
total_num_peers,
|
||||
"ip_version" => I::ip_version_str(),
|
||||
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
|
||||
);
|
||||
peer_gauge.set(total_num_peers);
|
||||
|
||||
torrent_map.shrink_to_fit();
|
||||
}
|
||||
|
|
@ -238,6 +271,7 @@ impl<I: Ip> TorrentData<I> {
|
|||
peer_ip_address: I,
|
||||
request: AnnounceRequest,
|
||||
valid_until: ValidUntil,
|
||||
#[cfg(feature = "metrics")] peer_gauge: &::metrics::Gauge,
|
||||
) -> (usize, usize, Vec<ResponsePeer<I>>) {
|
||||
let peer_status =
|
||||
PeerStatus::from_event_and_bytes_left(request.event, Some(request.bytes_left));
|
||||
|
|
@ -257,12 +291,7 @@ impl<I: Ip> TorrentData<I> {
|
|||
PeerStatus::Seeding | PeerStatus::Leeching => {
|
||||
#[cfg(feature = "metrics")]
|
||||
if opt_removed_peer.is_none() {
|
||||
::metrics::increment_gauge!(
|
||||
"aquatic_peers",
|
||||
1.0,
|
||||
"ip_version" => I::ip_version_str(),
|
||||
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
|
||||
);
|
||||
peer_gauge.increment(1.0);
|
||||
}
|
||||
|
||||
let max_num_peers_to_take = match request.numwant {
|
||||
|
|
@ -288,12 +317,7 @@ impl<I: Ip> TorrentData<I> {
|
|||
PeerStatus::Stopped => {
|
||||
#[cfg(feature = "metrics")]
|
||||
if opt_removed_peer.is_some() {
|
||||
::metrics::decrement_gauge!(
|
||||
"aquatic_peers",
|
||||
1.0,
|
||||
"ip_version" => I::ip_version_str(),
|
||||
"worker_index" => WORKER_INDEX.with(|index| index.get()).to_string(),
|
||||
);
|
||||
peer_gauge.decrement(1.0);
|
||||
}
|
||||
|
||||
Vec::new()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue