feat: allow configuring timeout

Cargo.lock

@@ -85,12 +85,47 @@ dependencies = [
  "ansi_term",
  "atty",
  "bitflags",
- "strsim",
+ "strsim 0.8.0",
  "textwrap",
  "unicode-width",
  "vec_map",
 ]
 
+[[package]]
+name = "darling"
+version = "0.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d0d720b8683f8dd83c65155f0530560cba68cd2bf395f6513a483caee57ff7f4"
+dependencies = [
+ "darling_core",
+ "darling_macro",
+]
+
+[[package]]
+name = "darling_core"
+version = "0.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7a340f241d2ceed1deb47ae36c4144b2707ec7dd0b649f894cb39bb595986324"
+dependencies = [
+ "fnv",
+ "ident_case",
+ "proc-macro2",
+ "quote",
+ "strsim 0.10.0",
+ "syn",
+]
+
+[[package]]
+name = "darling_macro"
+version = "0.13.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "72c41b3b7352feb3211a0d743dc5700a4e3b60f51bd2b368892d1e0f9a95f44b"
+dependencies = [
+ "darling_core",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "fern"
 version = "0.6.0"
@@ -301,6 +336,12 @@ dependencies = [
  "want",
 ]
 
+[[package]]
+name = "ident_case"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39"
+
 [[package]]
 name = "indexmap"
 version = "1.6.2"
@@ -464,6 +505,7 @@ dependencies = [
  "log",
  "prometheus",
  "serde",
+ "serde_with",
  "tokio",
  "tokio-icmp-echo",
  "toml",
@@ -607,6 +649,12 @@ dependencies = [
  "bitflags",
 ]
 
+[[package]]
+name = "rustversion"
+version = "1.0.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f2cc38e8fa666e2de3c4aba7edeb5ffc5246c1c2ed0e3d17e560aeeba736b23f"
+
 [[package]]
 name = "scopeguard"
 version = "1.1.0"
@@ -633,6 +681,29 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "serde_with"
+version = "1.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ec1e6ec4d8950e5b1e894eac0d360742f3b1407a6078a604a731c4b3f49cefbc"
+dependencies = [
+ "rustversion",
+ "serde",
+ "serde_with_macros",
+]
+
+[[package]]
+name = "serde_with_macros"
+version = "1.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12e47be9471c72889ebafb5e14d5ff930d89ae7a67bbdb5f8abb564f845a927e"
+dependencies = [
+ "darling",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "slab"
 version = "0.4.3"
@@ -661,6 +732,12 @@ version = "0.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8ea5119cdb4c55b55d432abb513a0429384878c15dde60cc77b1c99de1a95a6a"
 
+[[package]]
+name = "strsim"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"
+
 [[package]]
 name = "syn"
 version = "1.0.70"

Cargo.toml

@@ -20,6 +20,7 @@ fern = "0.6"
 log = { version = "0.4", features = ["serde"] }
 chrono = "0.4"
 serde = { version = "1", features = ["derive"] }
+serde_with = "1"
 tokio-icmp-echo = "0.4"
 futures-util = "0.3"
 anyhow = "1"

config.toml.sample

@@ -2,6 +2,11 @@
 # Takes the format IPv4:port or [IPv6]:port
 listener = "[::]:9898"
 
+# Configuration of peshmings pinging behaviour
+[ping]
+# The timeout is specified in milliseconds, with a default of 3 seconds.
+# timeout = 3000
+
 # Array of hosts to ping. Currently only supports plain IPs, no DNS names.
 # The format here is `"host" = interval in milliseconds`, so these examples
 # will ping the primary and secondary IP of cloudflare's 1.1.1.1 DNS service

src/config.rs

@@ -21,7 +21,8 @@ use anyhow::{Context, Result};
 use clap::{clap_app, crate_authors, crate_description, crate_name, crate_version};
 use log::info;
 use serde::Deserialize;
-use std::collections::HashMap;
+use serde_with::{serde_as, DurationMilliSeconds};
+use std::{collections::HashMap, time::Duration};
 
 #[derive(Deserialize, Clone)]
 pub(crate) struct Config {
@@ -31,11 +32,19 @@ pub(crate) struct Config {
     pub(crate) log: LogConfig,
 }
 
+#[serde_as]
 #[derive(Deserialize, Clone)]
 pub(crate) struct PingConfig {
+    #[serde_as(as = "DurationMilliSeconds<f64>")]
+    #[serde(default = "default_timeout")]
+    pub(crate) timeout: Duration,
     pub(crate) hosts: HashMap<std::net::IpAddr, u64>,
 }
 
+fn default_timeout() -> Duration {
+    Duration::from_secs(3)
+}
+
 #[derive(Deserialize, Clone)]
 pub(crate) struct LogConfig {
     pub(crate) level: log::LevelFilter,

src/ping.rs

@@ -46,15 +46,20 @@ pub(crate) async fn start_pinging_hosts(config: &Config) -> Result<()> {
     let mut handles = vec![];
     for (host, interval) in config.ping.hosts.clone() {
         info!("Spawn ping task for {}", host);
-        handles.push(tokio::spawn(ping_host(pinger.clone(), host, interval)));
+        handles.push(tokio::spawn(ping_host(
+            pinger.clone(),
+            host,
+            interval,
+            config.ping.timeout,
+        )));
     }
     let (result, _, _) = futures::future::select_all(handles).await;
     result??;
     Ok(())
 }
 
-async fn ping_host(pinger: Pinger, host: IpAddr, interval: u64) -> Result<()> {
+async fn ping_host(pinger: Pinger, host: IpAddr, interval: u64, timeout: Duration) -> Result<()> {
-    let mut pingchain = pinger.chain(host).timeout(Duration::from_secs(3));
+    let mut pingchain = pinger.chain(host).timeout(timeout);
     let mut interval = tokio::time::interval(Duration::from_millis(interval));
     let host_string = host.to_string();
     loop {
@@ -62,11 +67,12 @@ async fn ping_host(pinger: Pinger, host: IpAddr, interval: u64) -> Result<()> {
         tokio::spawn(catch(handle_ping_result(
             pingchain.send(),
             host_string.clone(),
+            timeout,
         )));
     }
 }
 
-async fn handle_ping_result(result: PingFuture, host: String) -> Result<()> {
+async fn handle_ping_result(result: PingFuture, host: String, timeout: Duration) -> Result<()> {
     let pong = result.await.context(format!("Couldn't ping {}", &host))?;
     match pong {
         Some(time) => {
@@ -78,7 +84,9 @@ async fn handle_ping_result(result: PingFuture, host: String) -> Result<()> {
         }
         None => {
             trace!("Received no response from {} within timeout", &host);
-            PING_HISTOGRAM.with_label_values(&[&host]).observe(3000.0);
+            PING_HISTOGRAM
+                .with_label_values(&[&host])
+                .observe(timeout.as_millis() as f64);
         }
     };