-
Notifications
You must be signed in to change notification settings - Fork 339
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
server: fix stats sending performance and ns creation (#1440)
* server: create debug logging * server: fix stats sending performance This change fixes how we send stats to the pulse server. This is done by not serially sending stats on the same task that we extract events from. The reason for this is that if the event buffer (stats_sender) gets full it will block namespace create requests. With this change, we now submit stats http requests off the main stats task and restrict the stats sending to 128 concurrent requests. This will allow us to accept more create namespace requests and efficiently send stats.
- Loading branch information
1 parent
97beabd
commit 2b9ad5c
Showing
7 changed files
with
108 additions
and
53 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,91 +1,104 @@ | ||
| #![allow(clippy::mutable_key_type)] | ||
|
|
||
| use std::collections::HashMap; | ||
| use std::sync::Weak; | ||
| use std::sync::{Arc, Weak}; | ||
| use std::time::Duration; | ||
| use tokio::time::Interval; | ||
| use url::Url; | ||
|
|
||
| use tokio::sync::mpsc; | ||
| use tokio::sync::{mpsc, Semaphore}; | ||
|
|
||
| use crate::http::admin::stats::StatsResponse; | ||
| use crate::namespace::{NamespaceName, NamespaceStore}; | ||
| use crate::namespace::meta_store::MetaStoreHandle; | ||
| use crate::namespace::NamespaceName; | ||
| use crate::stats::Stats; | ||
|
|
||
| pub async fn server_heartbeat( | ||
| url: Option<Url>, | ||
| auth: Option<String>, | ||
| update_period: Duration, | ||
| mut stats_subs: mpsc::Receiver<(NamespaceName, Weak<Stats>)>, | ||
| namespaces: NamespaceStore, | ||
| mut stats_subs: mpsc::Receiver<(NamespaceName, MetaStoreHandle, Weak<Stats>)>, | ||
| ) { | ||
| let mut watched = HashMap::new(); | ||
| let client = reqwest::Client::new(); | ||
| let mut interval = tokio::time::interval(update_period); | ||
| interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); | ||
|
|
||
| let semaphore = Arc::new(Semaphore::new(128)); | ||
|
|
||
| loop { | ||
| let wait_for_next_tick = next_tick(&mut interval, &semaphore, 128); | ||
|
|
||
| tokio::select! { | ||
| Some((ns, stats)) = stats_subs.recv() => { | ||
| watched.insert(ns, stats); | ||
| Some((ns, handle, stats)) = stats_subs.recv() => { | ||
| watched.insert(ns, (handle, stats)); | ||
| } | ||
| _ = interval.tick() => { | ||
| send_stats(&mut watched, &client, &namespaces, url.as_ref(), auth.as_deref()).await; | ||
| _ = wait_for_next_tick => { | ||
| send_stats(&mut watched, &client, url.as_ref(), auth.as_deref(), &semaphore).await; | ||
| } | ||
| }; | ||
| } | ||
| } | ||
|
|
||
| /// Wait for all the permits to be available again, this should work as long as its called after | ||
| /// the last `send_stats` is called since the sempaphore waits in a queue. | ||
| async fn next_tick(interval: &mut Interval, semaphore: &Arc<Semaphore>, permits: u32) { | ||
| let permit = semaphore.acquire_many(permits).await; | ||
| drop(permit); | ||
|
|
||
| interval.tick().await; | ||
| } | ||
|
|
||
| async fn send_stats( | ||
| watched: &mut HashMap<NamespaceName, Weak<Stats>>, | ||
| watched: &mut HashMap<NamespaceName, (MetaStoreHandle, Weak<Stats>)>, | ||
| client: &reqwest::Client, | ||
| namespaces: &NamespaceStore, | ||
| url: Option<&Url>, | ||
| auth: Option<&str>, | ||
| semaphore: &Arc<Semaphore>, | ||
| ) { | ||
| // first send all the stats... | ||
| for (ns, stats) in watched.iter() { | ||
| for (ns, (config_store, stats)) in watched.iter() { | ||
| if let Some(stats) = stats.upgrade() { | ||
| let body = StatsResponse::from(stats.as_ref()); | ||
| let mut heartbeat_url; | ||
| if let Some(url) = url { | ||
| heartbeat_url = url.clone(); | ||
|
|
||
| let mut heartbeat_url = if let Some(url) = url { | ||
| url.clone() | ||
| } else { | ||
| match namespaces.config_store(ns.clone()).await { | ||
| Ok(config_store) => { | ||
| let config = config_store.get(); | ||
| if let Some(url) = config.heartbeat_url.as_ref() { | ||
| heartbeat_url = url.clone(); | ||
| } else { | ||
| tracing::debug!( | ||
| "No heartbeat url for namespace {}. Can't send stats!", | ||
| ns.as_str() | ||
| ); | ||
| continue; | ||
| } | ||
| } | ||
| Err(e) => { | ||
| tracing::warn!( | ||
| "Error fetching config for namespace {}. Can't send stats: {}", | ||
| ns.as_str(), | ||
| e | ||
| ); | ||
| continue; | ||
| } | ||
| let config = config_store.get(); | ||
| if let Some(url) = config.heartbeat_url.as_ref() { | ||
| url.clone() | ||
| } else { | ||
| tracing::debug!( | ||
| "No heartbeat url for namespace {}. Can't send stats!", | ||
| ns.as_str() | ||
| ); | ||
| continue; | ||
| } | ||
| } | ||
| }; | ||
|
|
||
| heartbeat_url.path_segments_mut().unwrap().push(ns.as_str()); | ||
|
|
||
| let request = client.post(heartbeat_url); | ||
|
|
||
| let request = if let Some(ref auth) = auth { | ||
| request.header("Authorization", auth.to_string()) | ||
| } else { | ||
| request | ||
| }; | ||
|
|
||
| let request = request.json(&body); | ||
| if let Err(err) = request.send().await { | ||
| tracing::warn!("Error sending heartbeat: {}", err); | ||
| } | ||
|
|
||
| let semaphore = semaphore.clone(); | ||
| tokio::spawn(async move { | ||
| let _permit = semaphore.acquire().await; | ||
|
|
||
| if let Err(err) = request.send().await { | ||
| tracing::warn!("Error sending heartbeat: {}", err); | ||
| } | ||
| }); | ||
| } | ||
| } | ||
|
|
||
| // ..and then remove all expired subscription | ||
| watched.retain(|_, s| s.upgrade().is_some()); | ||
| watched.retain(|_, (_, s)| s.upgrade().is_some()); | ||
| } |
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
Oops, something went wrong.