chore: replace homegrown errors with anyhow

main
Jan Christian Grünhage 4 years ago
parent e6210b42c9
commit 82f7101f54

@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased] ## [Unreleased]
### Changed ### Changed
- update dependencies, including stable tokio this time. - update dependencies, including stable tokio this time.
- migrate error handling to anyhow
## [v0.3.0] - 2021-04-19 ## [v0.3.0] - 2021-04-19
### Added ### Added

19
Cargo.lock generated

@ -9,6 +9,23 @@ dependencies = [
"winapi", "winapi",
] ]
[[package]]
name = "anyhow"
version = "1.0.40"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "28b2cd92db5cbd74e8e5028f7e27dd7aa3090e89e4f2a197cc7c8dfb69c7063b"
[[package]]
name = "async-anyhow-logger"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06a2f0f9e176eb6d5185a049090385608c2294b14ef1647af489414521b0de1c"
dependencies = [
"anyhow",
"futures",
"log",
]
[[package]] [[package]]
name = "atty" name = "atty"
version = "0.2.14" version = "0.2.14"
@ -432,6 +449,8 @@ dependencies = [
name = "peshming" name = "peshming"
version = "0.3.0" version = "0.3.0"
dependencies = [ dependencies = [
"anyhow",
"async-anyhow-logger",
"chrono", "chrono",
"clap", "clap",
"fern", "fern",

@ -19,3 +19,5 @@ chrono = "0.4.19"
serde = { version = "1.0.125", features = ["derive"] } serde = { version = "1.0.125", features = ["derive"] }
tokio-ping = { git = "https://github.com/jcgruenhage/tokio-ping", branch = "main" } tokio-ping = { git = "https://github.com/jcgruenhage/tokio-ping", branch = "main" }
futures-util = "0.3.14" futures-util = "0.3.14"
anyhow = "1.0.40"
async-anyhow-logger = "0.1.0"

@ -17,6 +17,7 @@
* You should have received a copy of the GNU Affero General Public License * * You should have received a copy of the GNU Affero General Public License *
* along with this program. If not, see <https://www.gnu.org/licenses/>. * * along with this program. If not, see <https://www.gnu.org/licenses/>. *
********************************************************************************/ ********************************************************************************/
use anyhow::{Context, Result};
use clap::{clap_app, crate_authors, crate_description, crate_name, crate_version}; use clap::{clap_app, crate_authors, crate_description, crate_name, crate_version};
use log::info; use log::info;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@ -69,21 +70,7 @@ pub(crate) fn setup_fern(level: u64) {
} }
} }
pub(crate) fn read_config(path: &str) -> Result<Config, Error> { pub(crate) fn read_config(path: &str) -> Result<Config> {
let config_file_content = std::fs::read_to_string(path)?; let config_file_content = std::fs::read_to_string(path).context("Couldn't read config file")?;
Ok(toml::from_str(&config_file_content)?) Ok(toml::from_str(&config_file_content).context("Couldn't parse config file")?)
}
pub(crate) struct Error {}
impl std::convert::From<std::io::Error> for Error {
fn from(_: std::io::Error) -> Self {
Error {}
}
}
impl std::convert::From<toml::de::Error> for Error {
fn from(_: toml::de::Error) -> Self {
Error {}
}
} }

@ -17,33 +17,29 @@
* You should have received a copy of the GNU Affero General Public License * * You should have received a copy of the GNU Affero General Public License *
* along with this program. If not, see <https://www.gnu.org/licenses/>. * * along with this program. If not, see <https://www.gnu.org/licenses/>. *
********************************************************************************/ ********************************************************************************/
use log::error; use anyhow::{Context, Result};
mod config; mod config;
mod metrics; mod metrics;
mod ping; mod ping;
use crate::config::{read_config, setup_clap, setup_fern}; use crate::config::{read_config, setup_clap, setup_fern};
use crate::metrics::start_serving_metrics; use crate::metrics::start_serving_metrics;
use crate::ping::start_pinging_hosts; use crate::ping::start_pinging_hosts;
#[tokio::main] #[tokio::main]
async fn main() -> Result<(), ()> { async fn main() -> Result<()> {
let clap = setup_clap(); let clap = setup_clap();
setup_fern(clap.occurrences_of("v")); setup_fern(clap.occurrences_of("v"));
let config = match read_config(clap.value_of("config").unwrap()) { let config =
Ok(config) => config, read_config(clap.value_of("config").unwrap()).context("Couldn't read config file!")?;
Err(_) => {
error!("Couldn't read config file!"); let ping_fut = start_pinging_hosts(config.clone());
return Err(()); let serve_fut = start_serving_metrics(config.clone());
}
}; futures::pin_mut!(ping_fut);
futures::pin_mut!(serve_fut);
tokio::spawn(start_pinging_hosts(config.clone())); futures::future::select(ping_fut, serve_fut).await;
match start_serving_metrics(config.clone()).await { Ok(())
Ok(_) => Ok(()),
Err(error) => {
error!("Couldn't serve metrics: {}", error);
Err(())
}
}
} }

@ -18,6 +18,7 @@
* along with this program. If not, see <https://www.gnu.org/licenses/>. * * along with this program. If not, see <https://www.gnu.org/licenses/>. *
********************************************************************************/ ********************************************************************************/
use crate::config::Config; use crate::config::Config;
use anyhow::{Context, Result};
use hyper::{ use hyper::{
header::CONTENT_TYPE, header::CONTENT_TYPE,
server::Server, server::Server,
@ -50,7 +51,7 @@ lazy_static! {
.unwrap(); .unwrap();
} }
async fn serve_req(req: Request<Body>) -> std::result::Result<Response<Body>, hyper::Error> { async fn serve_req(req: Request<Body>) -> Result<Response<Body>> {
match req.uri().path() { match req.uri().path() {
"/metrics" => serve_metrics().await, "/metrics" => serve_metrics().await,
"/health" => serve_health_check().await, "/health" => serve_health_check().await,
@ -58,7 +59,7 @@ async fn serve_req(req: Request<Body>) -> std::result::Result<Response<Body>, hy
} }
} }
async fn serve_metrics() -> std::result::Result<Response<Body>, hyper::Error> { async fn serve_metrics() -> Result<Response<Body>> {
let encoder = TextEncoder::new(); let encoder = TextEncoder::new();
HTTP_COUNTER.inc(); HTTP_COUNTER.inc();
@ -73,31 +74,31 @@ async fn serve_metrics() -> std::result::Result<Response<Body>, hyper::Error> {
.status(200) .status(200)
.header(CONTENT_TYPE, encoder.format_type()) .header(CONTENT_TYPE, encoder.format_type())
.body(Body::from(buffer)) .body(Body::from(buffer))
.unwrap(); .context("Couldn't build metrics response")?;
timer.observe_duration(); timer.observe_duration();
Ok(response) Ok(response)
} }
async fn serve_health_check() -> std::result::Result<Response<Body>, hyper::Error> { async fn serve_health_check() -> Result<Response<Body>> {
Ok(Response::builder() Ok(Response::builder()
.status(200) .status(200)
.body(Body::from(vec![])) .body(Body::from(vec![]))
.unwrap()) .context("Couldn't build health check response")?)
} }
async fn serve_not_found() -> std::result::Result<Response<Body>, hyper::Error> { async fn serve_not_found() -> Result<Response<Body>> {
Ok(Response::builder() Ok(Response::builder()
.status(404) .status(404)
.body(Body::from(vec![])) .body(Body::from(vec![]))
.unwrap()) .context("Couldn't build not found response")?)
} }
pub(crate) async fn start_serving_metrics(config: Config) -> std::result::Result<(), hyper::Error> { pub(crate) async fn start_serving_metrics(config: Config) -> Result<()> {
let serve_future = Server::bind(&config.listener).serve(make_service_fn(|_| async { let serve_future = Server::bind(&config.listener).serve(make_service_fn(|_| async {
Ok::<_, hyper::Error>(service_fn(serve_req)) Ok::<_, hyper::Error>(service_fn(serve_req))
})); }));
info!("Listening on {}", &config.listener); info!("Listening on {}", &config.listener);
serve_future.await Ok(serve_future.await?)
} }

@ -18,8 +18,10 @@
* along with this program. If not, see <https://www.gnu.org/licenses/>. * * along with this program. If not, see <https://www.gnu.org/licenses/>. *
********************************************************************************/ ********************************************************************************/
use crate::config::Config; use crate::config::Config;
use anyhow::{Context, Result};
use async_anyhow_logger::catch;
use lazy_static::lazy_static; use lazy_static::lazy_static;
use log::{error, info, trace}; use log::{info, trace};
use prometheus::*; use prometheus::*;
use std::net::IpAddr; use std::net::IpAddr;
use std::time::Duration; use std::time::Duration;
@ -39,41 +41,33 @@ lazy_static! {
.unwrap(); .unwrap();
} }
pub(crate) async fn start_pinging_hosts( pub(crate) async fn start_pinging_hosts(config: Config) -> Result<()> {
config: Config, let pinger = Pinger::new().await.context("Couldn't create pinger")?;
) -> std::result::Result<(), tokio_ping::Error> { let mut handles = vec![];
let pinger = match Pinger::new().await {
Ok(pinger) => pinger,
Err(error) => {
error!("Couldn't create pinger: {}", error);
std::process::exit(1);
}
};
for (host, interval) in config.hosts.clone() { for (host, interval) in config.hosts.clone() {
info!("Spawn ping task for {}", host); info!("Spawn ping task for {}", host);
tokio::spawn(ping_host(pinger.clone(), host, interval)); handles.push(tokio::spawn(catch(ping_host(pinger.clone(), host, interval))));
} }
let (result, _, _) = futures::future::select_all(handles).await;
result?;
Ok(()) Ok(())
} }
async fn ping_host(pinger: Pinger, host: IpAddr, interval: u64) { async fn ping_host(pinger: Pinger, host: IpAddr, interval: u64) -> Result<()> {
let mut pingchain = pinger.chain(host).timeout(Duration::from_secs(3)); let mut pingchain = pinger.chain(host).timeout(Duration::from_secs(3));
let mut interval = tokio::time::interval(Duration::from_millis(interval)); let mut interval = tokio::time::interval(Duration::from_millis(interval));
let host_string = host.to_string(); let host_string = host.to_string();
loop { loop {
interval.tick().await; interval.tick().await;
tokio::spawn(handle_ping_result(pingchain.send(), host_string.clone())); tokio::spawn(catch(handle_ping_result(
pingchain.send(),
host_string.clone(),
)));
} }
} }
async fn handle_ping_result(result: PingFuture, host: String) { async fn handle_ping_result(result: PingFuture, host: String) -> Result<()> {
let pong = match result.await { let pong = result.await.context(format!("Couldn't ping {}", &host))?;
Ok(pong) => pong,
Err(error) => {
error!("Couldn't ping {}: {}", &host, error);
return;
}
};
match pong { match pong {
Some(time) => { Some(time) => {
let ms = time.as_millis(); let ms = time.as_millis();
@ -87,4 +81,6 @@ async fn handle_ping_result(result: PingFuture, host: String) {
PING_HISTOGRAM.with_label_values(&[&host]).observe(3000.0); PING_HISTOGRAM.with_label_values(&[&host]).observe(3000.0);
} }
}; };
Ok(())
} }

Loading…
Cancel
Save