finga/webhookey
1
0
Fork 0
Code Issues 5 Pull requests Projects Releases Wiki Activity

Use atomics instead of mutexes

Browse source 
Use atomics instead of mutexes to improve access on metrics. Therefor
also remove unneeded borrows.
This commit is contained in:
finga 2021-11-13 14:35:40 +01:00
parent b4b46ebd58
commit a122bf28d2
2 changed files with 38 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

93
src/main.rs
View file

@ -29,7 +29,7 @@ use std::{
fs::File, fs::File,
io::BufReader, io::BufReader,
net::{IpAddr, Ipv4Addr, SocketAddr}, net::{IpAddr, Ipv4Addr, SocketAddr},
sync::Mutex, sync::atomic::{AtomicUsize, Ordering},
}; };
mod cli; mod cli;
@ -42,15 +42,15 @@ use crate::{
#[derive(Debug, Default)] #[derive(Debug, Default)]
struct WebhookeyMetrics { struct WebhookeyMetrics {
requests_received: Mutex<usize>, requests_received: AtomicUsize,
requests_invalid: Mutex<usize>, requests_invalid: AtomicUsize,
hooks_successful: Mutex<usize>, hooks_successful: AtomicUsize,
hooks_forbidden: Mutex<usize>, hooks_forbidden: AtomicUsize,
hooks_unmatched: Mutex<usize>, hooks_unmatched: AtomicUsize,
commands_executed: Mutex<usize>, commands_executed: AtomicUsize,
commands_execution_failed: Mutex<usize>, commands_execution_failed: AtomicUsize,
commands_successful: Mutex<usize>, commands_successful: AtomicUsize,
commands_failed: Mutex<usize>, commands_failed: AtomicUsize,
} }
#[derive(Debug, Deserialize, Serialize)] #[derive(Debug, Deserialize, Serialize)]
@ -304,15 +304,15 @@ webhookey_commands_successful {}
# TYPE webhookey_commands_failed gauge # TYPE webhookey_commands_failed gauge
webhookey_commands_failed {} webhookey_commands_failed {}
", ",
metrics.requests_received.lock().unwrap(), // TODO: Check if unwrap need to be fixed metrics.requests_received.load(Ordering::Relaxed),
metrics.requests_invalid.lock().unwrap(), // TODO: Check if unwrap need to be fixed metrics.requests_invalid.load(Ordering::Relaxed),
metrics.hooks_successful.lock().unwrap(), // TODO: Check if unwrap need to be fixed metrics.hooks_successful.load(Ordering::Relaxed),
metrics.hooks_forbidden.lock().unwrap(), // TODO: Check if unwrap need to be fixed metrics.hooks_forbidden.load(Ordering::Relaxed),
metrics.hooks_unmatched.lock().unwrap(), // TODO: Check if unwrap need to be fixed metrics.hooks_unmatched.load(Ordering::Relaxed),
metrics.commands_executed.lock().unwrap(), // TODO: Check if unwrap need to be fixed metrics.commands_executed.load(Ordering::Relaxed),
metrics.commands_execution_failed.lock().unwrap(), // TODO: Check if unwrap need to be fixed metrics.commands_execution_failed.load(Ordering::Relaxed),
metrics.commands_successful.lock().unwrap(), // TODO: Check if unwrap need to be fixed metrics.commands_successful.load(Ordering::Relaxed),
metrics.commands_failed.lock().unwrap() // TODO: Check if unwrap need to be fixed metrics.commands_failed.load(Ordering::Relaxed),
) )
} }
@ -325,15 +325,12 @@ impl<'r> FromData<'r> for Hooks {
data: Data<'r>, data: Data<'r>,
) -> Outcome<Self, (Status, Self::Error), Data<'r>> { ) -> Outcome<Self, (Status, Self::Error), Data<'r>> {
{ {
let requests_received = &mut request request
.guard::<&State<WebhookeyMetrics>>() .guard::<&State<WebhookeyMetrics>>()
.await .await
.unwrap() // TODO: Check if unwrap need to be fixed .unwrap() // TODO: Check if unwrap need to be fixed
.requests_received .requests_received
.lock() .fetch_add(1, Ordering::Relaxed);
.unwrap(); // TODO: Check if unwrap need to be fixed
**requests_received += 1;
} }
match Hooks::get_commands(request, data).await { match Hooks::get_commands(request, data).await {
@ -343,15 +340,12 @@ impl<'r> FromData<'r> for Hooks {
.client_ip() .client_ip()
.unwrap_or(IpAddr::V4(Ipv4Addr::UNSPECIFIED)); .unwrap_or(IpAddr::V4(Ipv4Addr::UNSPECIFIED));
let hooks_unmatched = &mut request request
.guard::<&State<WebhookeyMetrics>>() .guard::<&State<WebhookeyMetrics>>()
.await .await
.unwrap() // TODO: Check if unwrap need to be fixed .unwrap() // TODO: Check if unwrap need to be fixed
.hooks_unmatched .hooks_unmatched
.lock() .fetch_add(1, Ordering::Relaxed);
.unwrap(); // TODO: Check if unwrap need to be fixed
**hooks_unmatched += 1;
warn!("Unmatched hook from {}", &client_ip); warn!("Unmatched hook from {}", &client_ip);
return Failure((Status::NotFound, WebhookeyError::UnmatchedHook(*client_ip))); return Failure((Status::NotFound, WebhookeyError::UnmatchedHook(*client_ip)));
@ -362,30 +356,24 @@ impl<'r> FromData<'r> for Hooks {
Err(WebhookeyError::Unauthorized(e)) => { Err(WebhookeyError::Unauthorized(e)) => {
error!("{}", WebhookeyError::Unauthorized(e)); error!("{}", WebhookeyError::Unauthorized(e));
let hooks_forbidden = &mut request request
.guard::<&State<WebhookeyMetrics>>() .guard::<&State<WebhookeyMetrics>>()
.await .await
.unwrap() // TODO: Check if unwrap need to be fixed .unwrap() // TODO: Check if unwrap need to be fixed
.hooks_forbidden .hooks_forbidden
.lock() .fetch_add(1, Ordering::Relaxed);
.unwrap(); // TODO: Check if unwrap need to be fixed
**hooks_forbidden += 1;
Failure((Status::Unauthorized, WebhookeyError::Unauthorized(e))) Failure((Status::Unauthorized, WebhookeyError::Unauthorized(e)))
} }
Err(e) => { Err(e) => {
error!("{}", e); error!("{}", e);
let requests_invalid = &mut request request
.guard::<&State<WebhookeyMetrics>>() .guard::<&State<WebhookeyMetrics>>()
.await .await
.unwrap() // TODO: Check if unwrap need to be fixed .unwrap() // TODO: Check if unwrap need to be fixed
.requests_invalid .requests_invalid
.lock() .fetch_add(1, Ordering::Relaxed);
.unwrap(); // TODO: Check if unwrap need to be fixed
**requests_invalid += 1;
Failure((Status::BadRequest, e)) Failure((Status::BadRequest, e))
} }
@ -410,26 +398,19 @@ async fn receive_hook<'a>(
trace!("Output of command `{}` on stdout: {:?}", &command, &stdout); trace!("Output of command `{}` on stdout: {:?}", &command, &stdout);
debug!("Output of command `{}` on stderr: {:?}", &command, &stderr); debug!("Output of command `{}` on stderr: {:?}", &command, &stderr);
let commands_executed = &mut metrics.commands_executed.lock().unwrap(); // TODO: Check if unwrap need to be fixed metrics.commands_executed.fetch_add(1, Ordering::Relaxed);
**commands_executed += 1;
match status { let _ = match status {
0 => { 0 => metrics.commands_successful.fetch_add(1, Ordering::Relaxed),
let commands_successful = &mut metrics.commands_successful.lock().unwrap(); // TODO: Check if unwrap need to be fixed _ => metrics.commands_failed.fetch_add(1, Ordering::Relaxed),
**commands_successful += 1; };
}
_ => {
let commands_failed = &mut metrics.commands_failed.lock().unwrap(); // TODO: Check if unwrap need to be fixed
**commands_failed += 1;
}
}
} }
Err(e) => { Err(e) => {
error!("Execution of `{}` failed: {}", &command, e); error!("Execution of `{}` failed: {}", &command, e);
let command_execution_failed = metrics
&mut metrics.commands_execution_failed.lock().unwrap(); // TODO: Check if unwrap need to be fixed .commands_execution_failed
**command_execution_failed += 1; .fetch_add(1, Ordering::Relaxed);
} }
} }
}); });
@ -446,10 +427,10 @@ async fn metrics(
if let Some(metrics_config) = &config.metrics { if let Some(metrics_config) = &config.metrics {
if let Some(filter) = &metrics_config.ip_filter { if let Some(filter) = &metrics_config.ip_filter {
if filter.validate(&address.ip()) { if filter.validate(&address.ip()) {
return Some(get_metrics(&metrics)); return Some(get_metrics(metrics));
} }
} else { } else {
return Some(get_metrics(&metrics)); return Some(get_metrics(metrics));
} }
} }

2
src/webhooks.rs
View file

@ -74,7 +74,7 @@ impl JsonFilter {
let regex = Regex::new(&self.regex).map_err(WebhookeyError::Regex)?; let regex = Regex::new(&self.regex).map_err(WebhookeyError::Regex)?;
if let Some(value) = data.pointer(&self.pointer) { if let Some(value) = data.pointer(&self.pointer) {
if regex.is_match(&self.get_string(&value)?) { if regex.is_match(&self.get_string(value)?) {
debug!("Regex `{}` for `{}` matches", &self.regex, &self.pointer); debug!("Regex `{}` for `{}` matches", &self.regex, &self.pointer);
return Ok(true); return Ok(true);