From 8c374133521b27e326fc332f170a01266c2d011c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Stian=20Fr=C3=B8ystein?= <26657174+stianfro@users.noreply.github.com> Date: Tue, 28 Jul 2026 01:56:40 +0000 Subject: [PATCH] feat: add local monitoring dashboard --- CHANGELOG.md | 7 + Cargo.lock | 163 +++++++++++ Cargo.toml | 7 +- README.md | 25 ++ deny.toml | 7 +- src/lib.rs | 768 +++++++++++++++++++++++++++++++++++++++++---------- 6 files changed, 825 insertions(+), 152 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 794c997..b5edc6a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ All notable changes to HTTPing are documented in this file. +## [Unreleased] + +### Added + +- A loopback-only live dashboard with summary cards, a latency chart, and recent probe results. +- Configurable dashboard ports, browser opening, and bounded in-memory history. + ## [0.1.0] - 2026-07-28 ### Added diff --git a/Cargo.lock b/Cargo.lock index 1f06293..8c6b9e7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -96,6 +96,54 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", +] + [[package]] name = "base64" version = "0.22.1" @@ -306,6 +354,12 @@ version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2cd50c473c80f6d7c3670a752354b8e569b1a7cbfdc0419ec88e5edad85e0dc7" +[[package]] +name = "futures-sink" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e34418ac499d6305c2fb5ad0ed2f6ac998c5f8ca209b4510f7f94242c647e307" + [[package]] name = "futures-task" version = "0.3.33" @@ -390,15 +444,26 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "httping" version = "0.1.0" dependencies = [ + "axum", "chrono", "clap", "humantime", + "open", "reqwest", + "serde", + "serde_json", "tokio", + "tokio-stream", ] [[package]] @@ -420,6 +485,7 @@ dependencies = [ "http", "http-body", "httparse", + "httpdate", "itoa", "pin-project-lite", "smallvec", @@ -598,6 +664,25 @@ version = "2.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" +[[package]] +name = "is-docker" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "928bae27f42bc99b60d9ac7334e3a21d10ad8f1835a4e12ec3ec0464765ed1b3" +dependencies = [ + "once_cell", +] + +[[package]] +name = "is-wsl" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "173609498df190136aa7dea1a91db051746d339e18476eed5ca40521f02d7aa5" +dependencies = [ + "is-docker", + "once_cell", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -704,12 +789,24 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "memchr" version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "mio" version = "1.2.2" @@ -742,6 +839,16 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "open" +version = "5.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a0b3d059e795d52b8a72fef45658620edd4d9c359b338564aa14391ffa511ed5" +dependencies = [ + "is-wsl", + "libc", +] + [[package]] name = "openssl-probe" version = "0.2.1" @@ -1081,6 +1188,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" dependencies = [ "serde_core", + "serde_derive", ] [[package]] @@ -1103,6 +1211,30 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + [[package]] name = "shlex" version = "2.0.1" @@ -1299,6 +1431,31 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", + "tokio-util", +] + +[[package]] +name = "tokio-util" +version = "0.7.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + [[package]] name = "tower" version = "0.5.3" @@ -1752,3 +1909,9 @@ dependencies = [ "quote", "syn 2.0.119", ] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/Cargo.toml b/Cargo.toml index 6906fa6..0603688 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,11 +17,16 @@ publish = false dist = true [dependencies] +axum = { version = "0.8.9", default-features = false, features = ["http1", "json", "tokio"] } chrono = { version = "0.4.45", default-features = false, features = ["clock"] } clap = "4.6.4" humantime = "2.4.0" +open = "5.4.0" reqwest = { version = "0.13.4", default-features = false, features = ["rustls"] } -tokio = { version = "1.53.1", features = ["macros", "net", "rt-multi-thread", "signal", "time"] } +serde = { version = "1.0.229", features = ["derive"] } +serde_json = "1.0.151" +tokio = { version = "1.53.1", features = ["macros", "net", "rt-multi-thread", "signal", "sync", "time"], default-features = false } +tokio-stream = { version = "0.1.19", default-features = false, features = ["sync"] } [dev-dependencies] tokio = { version = "1.53.1", features = ["io-util"] } diff --git a/README.md b/README.md index 1aefa51..0b4ae98 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,31 @@ Options: Durations accept values such as `250ms`, `1s`, and `2m`. +### Local dashboard + +Start the local dashboard for a target: + +```console +$ httping serve https://example.com +HTTPing https://example.com/ +resolved: 93.184.216.34:443 +dashboard: http://127.0.0.1:52143/ +``` + +HTTPing selects a free local port and opens the dashboard in the default +browser. The dashboard shows live summary cards, a latency chart, and recent +probe results. It binds only to `127.0.0.1`. + +Use `--no-open` to keep the browser closed, `--port` to select a fixed port, +and `--history` to set the number of recent results kept in memory: + +```sh +httping serve --no-open --port 8080 --history 500 https://example.com +``` + +The server stays in the foreground until Ctrl-C. Run it through a shell job, +systemd, or a container when it must continue in the background. + ```console $ httping --count 2 https://example.com HTTPing https://example.com/ diff --git a/deny.toml b/deny.toml index e01829c..7efcc18 100644 --- a/deny.toml +++ b/deny.toml @@ -18,7 +18,12 @@ ignore = [] multiple-versions = "deny" wildcards = "deny" deny = [] -skip = [] +# Serde 1 uses syn 3 while Tokio and URL's ICU dependencies still use syn 2. +# Keep this exact exception until the dependency graph converges. +skip = [ + { crate = "syn@2" }, + { crate = "syn@3" }, +] skip-tree = [] [licenses] diff --git a/src/lib.rs b/src/lib.rs index 1128c76..9d2582a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,22 +1,34 @@ -//! Command-line parsing and runtime behavior for `HTTPing`. +//! Command-line parsing, probing, and dashboard support for `HTTPing`. +use std::collections::VecDeque; +use std::convert::Infallible; use std::fmt::Write as _; use std::io::{self, Write}; use std::net::SocketAddr; use std::process::ExitCode; +use std::sync::{Arc, RwLock}; use std::time::{Duration, Instant}; +use axum::extract::State; +use axum::response::Html; +use axum::response::sse::{Event as SseEvent, KeepAlive, Sse}; +use axum::routing::get; +use axum::{Json, Router}; use chrono::{Local, SecondsFormat}; use clap::{Arg, ArgMatches, Command}; use reqwest::{Client, Url}; -use tokio::net::lookup_host; +use serde::Serialize; +use tokio::net::{TcpListener, lookup_host}; +use tokio::sync::{broadcast, oneshot}; use tokio::time::{Instant as TokioInstant, sleep_until, timeout}; +use tokio_stream::StreamExt as _; +use tokio_stream::wrappers::BroadcastStream; /// Ping an HTTP endpoint and report response latency. #[derive(Debug)] pub struct Cli { - /// Number of probes to send. By default, probes continue until Ctrl-C. - pub count: Option, + /// Selected operating mode. + pub mode: Mode, /// Minimum time between probe starts. pub interval: Duration, @@ -28,6 +40,25 @@ pub struct Cli { pub url: Url, } +/// `HTTPing` operating mode. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Mode { + /// Write probe results to the terminal. + Probe { + /// Number of probes to send. By default, probes continue until Ctrl-C. + count: Option, + }, + /// Run the local web dashboard until Ctrl-C. + Serve { + /// Local dashboard port. Zero asks the operating system to select a port. + port: u16, + /// Do not open the dashboard in the default browser. + no_open: bool, + /// Maximum number of detailed probe events kept in memory. + history: usize, + }, +} + impl Cli { /// Parse the process command line, printing an error and exiting on invalid input. #[must_use] @@ -47,61 +78,120 @@ impl Cli { } fn from_matches(matches: &ArgMatches) -> Self { + if let Some(("serve", serve)) = matches.subcommand() { + return Self { + mode: Mode::Serve { + port: *serve.get_one::("port").expect("port has a default"), + no_open: serve.get_flag("no-open"), + history: *serve + .get_one::("history") + .expect("history has a default"), + }, + interval: required_value(serve, "interval"), + timeout: required_value(serve, "timeout"), + url: required_value::(serve, "url").clone(), + }; + } + Self { - count: matches.get_one::("count").copied(), - interval: *matches - .get_one::("interval") - .expect("interval has a default"), - timeout: *matches - .get_one::("timeout") - .expect("timeout has a default"), - url: matches - .get_one::("url") - .expect("URL is required") - .clone(), + mode: Mode::Probe { + count: matches.get_one::("count").copied(), + }, + interval: required_value(matches, "interval"), + timeout: required_value(matches, "timeout"), + url: required_value::(matches, "url").clone(), } } } +fn required_value(matches: &ArgMatches, name: &str) -> T { + matches + .get_one::(name) + .unwrap_or_else(|| panic!("{name} is required or has a default")) + .clone() +} + fn command() -> Command { Command::new("httping") .version(env!("CARGO_PKG_VERSION")) .about("Ping an HTTP endpoint and report response latency") - .arg( - Arg::new("count") - .short('c') - .long("count") - .value_name("COUNT") - .help("Number of probes to send. By default, probes continue until Ctrl-C") - .value_parser(clap::value_parser!(u64).range(1..)), - ) - .arg( - Arg::new("interval") - .short('i') - .long("interval") - .value_name("DURATION") - .help("Minimum time between probe starts") - .default_value("1s") - .value_parser(parse_duration), - ) - .arg( - Arg::new("timeout") - .short('t') - .long("timeout") - .value_name("DURATION") - .help("Maximum time for DNS setup and for each probe") - .default_value("10s") - .value_parser(parse_duration), - ) - .arg( - Arg::new("url") - .value_name("URL") - .help("HTTP or HTTPS URL to probe") - .required(true) - .value_parser(parse_url), + .subcommand_negates_reqs(true) + .args(probe_arguments()) + .subcommand( + Command::new("serve") + .about("Show live probe statistics in a local web dashboard") + .args(common_arguments()) + .arg( + Arg::new("port") + .long("port") + .value_name("PORT") + .help("Local dashboard port. By default, the operating system selects one") + .default_value("0") + .value_parser(clap::value_parser!(u16)), + ) + .arg( + Arg::new("no-open") + .long("no-open") + .help("Do not open the dashboard in the default browser") + .action(clap::ArgAction::SetTrue), + ) + .arg( + Arg::new("history") + .long("history") + .value_name("COUNT") + .help("Number of recent probe results kept in memory") + .default_value("300") + .value_parser(parse_positive_usize), + ), ) } +fn probe_arguments() -> Vec { + let mut arguments = common_arguments(); + arguments.insert( + 0, + Arg::new("count") + .short('c') + .long("count") + .value_name("COUNT") + .help("Number of probes to send. By default, probes continue until Ctrl-C") + .value_parser(clap::value_parser!(u64).range(1..)), + ); + arguments +} + +fn common_arguments() -> Vec { + vec![ + Arg::new("interval") + .short('i') + .long("interval") + .value_name("DURATION") + .help("Minimum time between probe starts") + .default_value("1s") + .value_parser(parse_duration), + Arg::new("timeout") + .short('t') + .long("timeout") + .value_name("DURATION") + .help("Maximum time for DNS setup and for each probe") + .default_value("10s") + .value_parser(parse_duration), + Arg::new("url") + .value_name("URL") + .help("HTTP or HTTPS URL to probe") + .required(true) + .value_parser(parse_url), + ] +} + +fn parse_positive_usize(value: &str) -> Result { + let count = value.parse::().map_err(|error| error.to_string())?; + if count == 0 { + return Err("count must be greater than zero".to_owned()); + } + Ok(count) +} + fn parse_duration(value: &str) -> Result { let duration = humantime::parse_duration(value).map_err(|error| error.to_string())?; if duration.is_zero() { @@ -118,43 +208,108 @@ fn parse_url(value: &str) -> Result { Ok(url) } -#[derive(Default)] +/// One completed HTTP probe. +#[derive(Clone, Debug, Serialize)] +pub struct ProbeEvent { + /// Increasing sequence number for this run. + pub sequence: u64, + /// Probe start time in RFC 3339 format. + pub timestamp: String, + /// Time from request start until response headers or failure. + pub elapsed_ms: f64, + /// Result of the probe. + #[serde(flatten)] + pub outcome: ProbeOutcome, +} + +/// Result data for a completed probe. +#[derive(Clone, Debug, Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum ProbeOutcome { + /// The endpoint returned HTTP response headers. + Response { + /// Numeric HTTP status code. + status: u16, + /// Display form of the HTTP status. + status_text: String, + /// Negotiated HTTP protocol version. + protocol_version: String, + }, + /// The request failed before response headers arrived. + Error { + /// Stable, low-cardinality error category. + kind: ErrorKind, + /// Human-readable error details. + message: String, + }, +} + +/// Stable probe transport error category. +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ErrorKind { + /// The probe exceeded its configured timeout. + Timeout, + /// The connection could not be established. + Connect, + /// A redirect could not be followed. + Redirect, + /// A request construction or transmission error occurred. + Request, + /// Another transport error occurred. + Other, +} + +#[derive(Clone, Debug, Default, Serialize)] struct Statistics { attempts: u64, responses: u64, + loss_percent: f64, + minimum_ms: Option, + average_ms: Option, + maximum_ms: Option, + latest_ms: Option, + #[serde(skip)] total: Duration, + #[serde(skip)] minimum: Option, + #[serde(skip)] maximum: Option, + #[serde(skip)] failed: bool, } impl Statistics { - fn response(&mut self, elapsed: Duration) { - self.attempts += 1; - self.responses += 1; - self.total += elapsed; - self.minimum = Some(self.minimum.map_or(elapsed, |value| value.min(elapsed))); - self.maximum = Some(self.maximum.map_or(elapsed, |value| value.max(elapsed))); - } - - fn error(&mut self) { + fn record(&mut self, event: &ProbeEvent) { self.attempts += 1; - self.failed = true; + let elapsed = Duration::from_secs_f64(event.elapsed_ms / 1_000.0); + self.latest_ms = Some(event.elapsed_ms); + if matches!(event.outcome, ProbeOutcome::Response { .. }) { + self.responses += 1; + self.total += elapsed; + self.minimum = Some(self.minimum.map_or(elapsed, |value| value.min(elapsed))); + self.maximum = Some(self.maximum.map_or(elapsed, |value| value.max(elapsed))); + self.minimum_ms = self.minimum.map(duration_millis); + self.maximum_ms = self.maximum.map(duration_millis); + self.average_ms = Some(duration_millis( + self.total / u32::try_from(self.responses).unwrap_or(u32::MAX), + )); + } else { + self.failed = true; + } + self.loss_percent = if self.attempts == 0 { + 0.0 + } else { + let loss_tenths = + (u128::from(self.attempts - self.responses) * 1_000) / u128::from(self.attempts); + f64::from(u32::try_from(loss_tenths).unwrap_or(1_000)) / 10.0 + }; } fn summary(&self) -> String { - let lost_attempts = self.attempts - self.responses; - let loss_tenths = if self.attempts == 0 { - 0 - } else { - (u128::from(lost_attempts) * 1_000) / u128::from(self.attempts) - }; let mut output = format!( - "\n--- httping statistics ---\nattempts: {}, responses: {}, loss: {}.{}%\n", - self.attempts, - self.responses, - loss_tenths / 10, - loss_tenths % 10 + "\n--- httping statistics ---\nattempts: {}, responses: {}, loss: {:.1}%\n", + self.attempts, self.responses, self.loss_percent ); if let (Some(minimum), Some(maximum)) = (self.minimum, self.maximum) { let average = self.total / u32::try_from(self.responses).unwrap_or(u32::MAX); @@ -172,22 +327,23 @@ impl Statistics { } } -/// Run `HTTPing` and return its process exit status. -pub async fn run(cli: Cli) -> ExitCode { - let stdout = io::stdout(); - let stderr = io::stderr(); - run_with_writers(cli, &mut stdout.lock(), &mut stderr.lock()).await +fn duration_millis(duration: Duration) -> f64 { + duration.as_secs_f64() * 1_000.0 } -#[allow(clippy::too_many_lines)] -async fn run_with_writers(cli: Cli, output: &mut impl Write, errors: &mut impl Write) -> ExitCode { +struct Prepared { + client: Client, + addresses: Vec, +} + +async fn prepare(cli: &Cli, errors: &mut impl Write) -> Result { let Some(host) = cli.url.host_str() else { let _ = writeln!(errors, "httping: URL has no hostname"); - return ExitCode::FAILURE; + return Err(ExitCode::FAILURE); }; let Some(port) = cli.url.port_or_known_default() else { let _ = writeln!(errors, "httping: URL has no effective port"); - return ExitCode::FAILURE; + return Err(ExitCode::FAILURE); }; let addresses = match timeout(cli.timeout, lookup_host((host, port))).await { @@ -197,11 +353,11 @@ async fn run_with_writers(cli: Cli, output: &mut impl Write, errors: &mut impl W "httping: DNS setup timed out after {}", humantime::format_duration(cli.timeout) ); - return ExitCode::FAILURE; + return Err(ExitCode::FAILURE); } Ok(Err(error)) => { let _ = writeln!(errors, "httping: could not resolve {host}:{port}: {error}"); - return ExitCode::FAILURE; + return Err(ExitCode::FAILURE); } Ok(Ok(addresses)) => { let mut addresses: Vec = addresses.collect(); @@ -212,42 +368,81 @@ async fn run_with_writers(cli: Cli, output: &mut impl Write, errors: &mut impl W errors, "httping: DNS returned no addresses for {host}:{port}" ); - return ExitCode::FAILURE; + return Err(ExitCode::FAILURE); } addresses } }; - let client = match Client::builder() + let client = Client::builder() .user_agent(concat!("httping/", env!("CARGO_PKG_VERSION"))) .build() - { - Ok(client) => client, - Err(error) => { + .map_err(|error| { let _ = writeln!(errors, "httping: could not create HTTP client: {error}"); - return ExitCode::FAILURE; - } - }; + ExitCode::FAILURE + })?; + + Ok(Prepared { client, addresses }) +} +fn write_header(cli: &Cli, prepared: &Prepared, output: &mut impl Write) { let _ = writeln!(output, "HTTPing {}", cli.url); - for address in &addresses { + for address in &prepared.addresses { let _ = writeln!(output, "resolved: {address}"); } +} + +fn write_event(event: &ProbeEvent, output: &mut impl Write) { + match &event.outcome { + ProbeOutcome::Response { status_text, .. } => { + let _ = writeln!( + output, + "{} {} {}", + event.timestamp, + status_text, + humantime::format_duration(Duration::from_secs_f64(event.elapsed_ms / 1_000.0)) + ); + } + ProbeOutcome::Error { message, .. } => { + let _ = writeln!( + output, + "{} error: {} {}", + event.timestamp, + message, + humantime::format_duration(Duration::from_secs_f64(event.elapsed_ms / 1_000.0)) + ); + } + } +} + +struct RunResult { + statistics: Statistics, + interrupted: bool, +} +async fn probe_loop( + cli: &Cli, + prepared: &Prepared, + count: Option, + mut observe: impl FnMut(&ProbeEvent), + errors: &mut impl Write, +) -> RunResult { let mut statistics = Statistics::default(); let mut next_start = TokioInstant::now(); + let mut interrupted = false; loop { - if cli.count.is_some_and(|count| statistics.attempts >= count) { + if count.is_some_and(|limit| statistics.attempts >= limit) { break; } - tokio::select! { () = sleep_until(next_start) => {} signal = tokio::signal::ctrl_c() => { if let Err(error) = signal { let _ = writeln!(errors, "httping: could not listen for Ctrl-C: {error}"); statistics.failed = true; + } else { + interrupted = true; } break; } @@ -256,71 +451,288 @@ async fn run_with_writers(cli: Cli, output: &mut impl Write, errors: &mut impl W let probe_start = Instant::now(); next_start = TokioInstant::now() + cli.interval; let timestamp = Local::now().to_rfc3339_opts(SecondsFormat::Millis, true); - let request = client.get(cli.url.clone()).send(); - - tokio::select! { - result = timeout(cli.timeout, request) => { - let elapsed = probe_start.elapsed(); - match result { - Ok(Ok(response)) => { - statistics.response(elapsed); - let _ = writeln!( - output, - "{} {} {}", - timestamp, - response.status(), - humantime::format_duration(elapsed) - ); - } - Ok(Err(error)) => { - statistics.error(); - let _ = writeln!( - output, - "{} error: {} {}", - timestamp, - error, - humantime::format_duration(elapsed) - ); - } - Err(_) => { - statistics.error(); - let _ = writeln!( - output, - "{} error: timed out after {} {}", - timestamp, - humantime::format_duration(cli.timeout), - humantime::format_duration(elapsed) - ); - } - } - } + let request = prepared.client.get(cli.url.clone()).send(); + let result = tokio::select! { + result = timeout(cli.timeout, request) => Some(result), signal = tokio::signal::ctrl_c() => { if let Err(error) = signal { let _ = writeln!(errors, "httping: could not listen for Ctrl-C: {error}"); statistics.failed = true; + } else { + interrupted = true; } - break; + None } - } + }; + let Some(result) = result else { break }; + let elapsed = probe_start.elapsed(); + let outcome = match result { + Ok(Ok(response)) => ProbeOutcome::Response { + status: response.status().as_u16(), + status_text: response.status().to_string(), + protocol_version: format!("{:?}", response.version()), + }, + Ok(Err(error)) => ProbeOutcome::Error { + kind: classify_error(&error), + message: error.to_string(), + }, + Err(_) => ProbeOutcome::Error { + kind: ErrorKind::Timeout, + message: format!( + "timed out after {}", + humantime::format_duration(cli.timeout) + ), + }, + }; + let event = ProbeEvent { + sequence: statistics.attempts + 1, + timestamp, + elapsed_ms: duration_millis(elapsed), + outcome, + }; + statistics.record(&event); + observe(&event); } - let _ = write!(output, "{}", statistics.summary()); - if cli.count.is_some() && statistics.failed { - ExitCode::FAILURE + RunResult { + statistics, + interrupted, + } +} + +fn classify_error(error: &reqwest::Error) -> ErrorKind { + if error.is_timeout() { + ErrorKind::Timeout + } else if error.is_connect() { + ErrorKind::Connect + } else if error.is_redirect() { + ErrorKind::Redirect + } else if error.is_request() { + ErrorKind::Request } else { - ExitCode::SUCCESS + ErrorKind::Other + } +} + +/// Run `HTTPing` and return its process exit status. +pub async fn run(cli: Cli) -> ExitCode { + let stdout = io::stdout(); + let stderr = io::stderr(); + run_with_writers(cli, &mut stdout.lock(), &mut stderr.lock()).await +} + +async fn run_with_writers(cli: Cli, output: &mut impl Write, errors: &mut impl Write) -> ExitCode { + let prepared = match prepare(&cli, errors).await { + Ok(prepared) => prepared, + Err(status) => return status, + }; + write_header(&cli, &prepared, output); + + match cli.mode { + Mode::Probe { count } => { + let result = probe_loop( + &cli, + &prepared, + count, + |event| write_event(event, output), + errors, + ) + .await; + let _ = write!(output, "{}", result.statistics.summary()); + if count.is_some() && result.statistics.failed { + ExitCode::FAILURE + } else { + ExitCode::SUCCESS + } + } + Mode::Serve { + port, + no_open, + history, + } => run_dashboard(&cli, &prepared, port, no_open, history, output, errors).await, + } +} + +#[derive(Clone, Debug, Serialize)] +struct DashboardState { + target: String, + started_at: String, + interval_ms: f64, + timeout_ms: f64, + resolved_addresses: Vec, + statistics: Statistics, + history: VecDeque, + last_sequence: u64, + history_limit: usize, +} + +impl DashboardState { + fn new(cli: &Cli, prepared: &Prepared, history_limit: usize) -> Self { + Self { + target: safe_target_name(&cli.url), + started_at: Local::now().to_rfc3339_opts(SecondsFormat::Millis, true), + interval_ms: duration_millis(cli.interval), + timeout_ms: duration_millis(cli.timeout), + resolved_addresses: prepared.addresses.iter().map(ToString::to_string).collect(), + statistics: Statistics::default(), + history: VecDeque::with_capacity(history_limit.min(4_096)), + last_sequence: 0, + history_limit, + } + } + + fn record(&mut self, event: &ProbeEvent) { + self.statistics.record(event); + self.last_sequence = event.sequence; + self.history.push_back(event.clone()); + while self.history.len() > self.history_limit { + self.history.pop_front(); + } + } +} + +#[derive(Clone)] +struct WebState { + dashboard: Arc>, + events: broadcast::Sender, +} + +async fn dashboard_page() -> Html<&'static str> { + Html(DASHBOARD_HTML) +} + +async fn dashboard_state(State(state): State) -> Json { + Json( + state + .dashboard + .read() + .expect("dashboard lock is not poisoned") + .clone(), + ) +} + +async fn dashboard_events( + State(state): State, +) -> Sse>> { + let stream = BroadcastStream::new(state.events.subscribe()).filter_map(|result| { + result.ok().and_then(|event| { + serde_json::to_string(&event).ok().map(|data| { + Ok(SseEvent::default() + .id(event.sequence.to_string()) + .event("probe") + .data(data)) + }) + }) + }); + Sse::new(stream).keep_alive(KeepAlive::default()) +} + +async fn run_dashboard( + cli: &Cli, + prepared: &Prepared, + port: u16, + no_open: bool, + history: usize, + output: &mut impl Write, + errors: &mut impl Write, +) -> ExitCode { + let listener = match TcpListener::bind(("127.0.0.1", port)).await { + Ok(listener) => listener, + Err(error) => { + let _ = writeln!(errors, "httping: could not bind dashboard: {error}"); + return ExitCode::FAILURE; + } + }; + let address = match listener.local_addr() { + Ok(address) => address, + Err(error) => { + let _ = writeln!(errors, "httping: could not read dashboard address: {error}"); + return ExitCode::FAILURE; + } + }; + let dashboard = Arc::new(RwLock::new(DashboardState::new(cli, prepared, history))); + let (events, _) = broadcast::channel(history.clamp(16, 4_096)); + let web_state = WebState { + dashboard: Arc::clone(&dashboard), + events: events.clone(), + }; + let app = Router::new() + .route("/", get(dashboard_page)) + .route("/api/state", get(dashboard_state)) + .route("/api/events", get(dashboard_events)) + .with_state(web_state); + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + let server = tokio::spawn(async move { + axum::serve(listener, app) + .with_graceful_shutdown(async move { + let _ = shutdown_rx.await; + }) + .await + }); + + let url = format!("http://{address}/"); + let _ = writeln!(output, "dashboard: {url}"); + if !no_open && open::that_detached(&url).is_err() { + let _ = writeln!(errors, "httping: could not open the dashboard in a browser"); } + + let result = probe_loop( + cli, + prepared, + None, + |event| { + write_event(event, output); + dashboard + .write() + .expect("dashboard lock is not poisoned") + .record(event); + let _ = events.send(event.clone()); + }, + errors, + ) + .await; + let _ = shutdown_tx.send(()); + match server.await { + Ok(Ok(())) => {} + Ok(Err(error)) => { + let _ = writeln!(errors, "httping: dashboard server failed: {error}"); + } + Err(error) => { + let _ = writeln!(errors, "httping: dashboard task failed: {error}"); + } + } + let _ = write!(output, "{}", result.statistics.summary()); + let _ = result.interrupted; + ExitCode::SUCCESS } +fn safe_target_name(url: &Url) -> String { + let host = url.host_str().unwrap_or("unknown"); + let port = url.port_or_known_default().unwrap_or(0); + format!("{}://{host}:{port}{}", url.scheme(), url.path()) +} + +const DASHBOARD_HTML: &str = r#" + +HTTPing dashboard
Connecting

HTTPing

Latency

Recent probes

#TimeResultLatency
+"#; + #[cfg(test)] mod tests { use super::*; - use clap::error::ErrorKind; + use clap::error::ErrorKind as ClapErrorKind; #[test] fn parses_defaults_and_duration_forms() { let cli = Cli::try_parse_from(["httping", "https://example.com"]).unwrap(); - assert_eq!(cli.count, None); + assert_eq!(cli.mode, Mode::Probe { count: None }); assert_eq!(cli.interval, Duration::from_secs(1)); assert_eq!(cli.timeout, Duration::from_secs(10)); @@ -335,11 +747,40 @@ mod tests { "http://example.com", ]) .unwrap(); - assert_eq!(cli.count, Some(2)); + assert_eq!(cli.mode, Mode::Probe { count: Some(2) }); assert_eq!(cli.interval, Duration::from_millis(250)); assert_eq!(cli.timeout, Duration::from_mins(2)); } + #[test] + fn parses_dashboard_options() { + let cli = Cli::try_parse_from([ + "httping", + "serve", + "--port", + "8080", + "--no-open", + "--history", + "10", + "--interval", + "250ms", + "https://example.com/private?token=secret", + ]) + .unwrap(); + assert_eq!( + cli.mode, + Mode::Serve { + port: 8080, + no_open: true, + history: 10 + } + ); + assert_eq!( + safe_target_name(&cli.url), + "https://example.com:443/private" + ); + } + #[test] fn rejects_bad_command_lines() { for arguments in [ @@ -347,36 +788,66 @@ mod tests { vec!["httping", "ftp://example.com"], vec!["httping", "not-a-url"], vec!["httping", "https://example.com", "extra"], + vec!["httping", "serve", "--history", "0", "https://example.com"], ] { assert_ne!( Cli::try_parse_from(arguments).unwrap_err().kind(), - ErrorKind::DisplayHelp + ClapErrorKind::DisplayHelp ); } } + #[test] + fn dashboard_history_is_bounded_but_statistics_are_not() { + let cli = Cli::try_parse_from(["httping", "https://example.com"]).unwrap(); + let prepared = Prepared { + client: Client::new(), + addresses: vec!["127.0.0.1:443".parse().unwrap()], + }; + let mut state = DashboardState::new(&cli, &prepared, 2); + for (sequence, elapsed_ms) in [(1, 1.0), (2, 2.0), (3, 3.0)] { + state.record(&ProbeEvent { + sequence, + timestamp: "2026-01-01T00:00:00Z".to_owned(), + elapsed_ms, + outcome: ProbeOutcome::Response { + status: 200, + status_text: "200 OK".to_owned(), + protocol_version: "HTTP/1.1".to_owned(), + }, + }); + } + assert_eq!(state.statistics.attempts, 3); + assert_eq!(state.history.len(), 2); + assert_eq!(state.history.front().unwrap().sequence, 2); + } + #[test] fn summary_handles_responses_and_total_loss() { let mut statistics = Statistics::default(); - statistics.response(Duration::from_millis(10)); - statistics.response(Duration::from_millis(30)); + for (sequence, elapsed_ms) in [(1, 10.0), (2, 30.0)] { + statistics.record(&ProbeEvent { + sequence, + timestamp: String::new(), + elapsed_ms, + outcome: ProbeOutcome::Response { + status: 200, + status_text: "200 OK".to_owned(), + protocol_version: "HTTP/1.1".to_owned(), + }, + }); + } assert!( statistics .summary() .contains("attempts: 2, responses: 2, loss: 0.0%") ); assert!(statistics.summary().contains("avg 20ms")); - - let mut statistics = Statistics::default(); - statistics.error(); - assert!(statistics.summary().contains("loss: 100.0%")); - assert!(statistics.summary().contains("min n/a")); } #[tokio::test] async fn counts_non_successful_http_responses_and_times_header_wait() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; - use tokio::net::TcpListener; let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); @@ -421,8 +892,6 @@ mod tests { #[tokio::test] async fn continues_after_timeouts_and_returns_failure_for_finite_run() { - use tokio::net::TcpListener; - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let server = tokio::spawn(async move { @@ -431,7 +900,6 @@ mod tests { tokio::time::sleep(Duration::from_millis(30)).await; } }); - let cli = Cli::try_parse_from([ "httping", "-c",