rexaping/src/ping.rs

/********************************************************************************
 *   Prometheus exporter for monitoring network connectivity using icmp pings   *
 *                                                                              *
 *   Copyright (C) 2019-2020 Jan Christian Grünhage                             *
 *   Copyright (C) 2020 Famedly GmbH                                            *
 *                                                                              *
 *   This program is free software: you can redistribute it and/or modify       *
 *   it under the terms of the GNU Affero General Public License as             *
 *   published by the Free Software Foundation, either version 3 of the         *
 *   License, or (at your option) any later version.                            *
 *                                                                              *
 *   This program is distributed in the hope that it will be useful,            *
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of             *
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the               *
 *   GNU Affero General Public License for more details.                        *
 *                                                                              *
 *   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/>.     *
 ********************************************************************************/
use crate::config::Config;
use anyhow::{Context, Result};
use async_anyhow_logger::catch;
use lazy_static::lazy_static;
use log::{info, trace};
use prometheus::*;
use std::net::IpAddr;
use std::time::Duration;
use tokio_ping::{PingFuture, Pinger};

lazy_static! {
    static ref PING_HISTOGRAM: HistogramVec = register_histogram_vec!(
        "ping_rtt_milliseconds",
        "The ping round trip time in milliseconds",
        &["target"],
        vec![
            0.5, 1.0, 5.0, 10.0, 15.0, 20.0, 25.0, 50.0, 75.0, 100.0, 150.0, 200.0, 250.0, 300.0,
            350.0, 400.0, 450.0, 500.0, 550.0, 600.0, 650.0, 700.0, 750.0, 800.0, 900.0, 1000.0,
            1250.0, 1500.0, 1750.0, 2000.0
        ]
    )
    .unwrap();
}

pub(crate) async fn start_pinging_hosts(config: &Config) -> Result<()> {
    let pinger = Pinger::new().await.context("Couldn't create pinger")?;
    let mut handles = vec![];
    for (host, interval) in config.hosts.clone() {
        info!("Spawn ping task for {}", host);
        handles.push(tokio::spawn(ping_host(pinger.clone(), host, interval)));
    }
    let (result, _, _) = futures::future::select_all(handles).await;
    result??;
    Ok(())
}

async fn ping_host(pinger: Pinger, host: IpAddr, interval: u64) -> Result<()> {
    let mut pingchain = pinger.chain(host).timeout(Duration::from_secs(3));
    let mut interval = tokio::time::interval(Duration::from_millis(interval));
    let host_string = host.to_string();
    loop {
        interval.tick().await;
        tokio::spawn(catch(handle_ping_result(
            pingchain.send(),
            host_string.clone(),
        )));
    }
}

async fn handle_ping_result(result: PingFuture, host: String) -> Result<()> {
    let pong = result.await.context(format!("Couldn't ping {}", &host))?;
    match pong {
        Some(time) => {
            let ms = time.as_millis();
            trace!("Received pong from {} after {} ms", &host, &ms);
            PING_HISTOGRAM
                .with_label_values(&[&host])
                .observe(ms as f64);
        }
        None => {
            trace!("Received no response from {} within timeout", &host);
            PING_HISTOGRAM.with_label_values(&[&host]).observe(3000.0);
        }
    };

    Ok(())
}