Merge pull request #39 from greatest-ape/better-udp-statistics

udp: support writing statistics html report to file

.github/workflows/cargo-build-and-test.yml

@@ -12,7 +12,7 @@ env:
 jobs:
   build-test-linux:
     runs-on: ubuntu-latest
-    timeout-minutes: 10
+    timeout-minutes: 20
     steps:
     - uses: actions/checkout@v2
     - name: Install dependencies
@@ -37,4 +37,4 @@ jobs:
     - name: Build
       run: |
         cargo build --verbose -p aquatic_udp
-        cargo build --verbose -p aquatic_ws
+        cargo build --verbose -p aquatic_ws

Cargo.lock

@@ -163,11 +163,13 @@ dependencies = [
  "aquatic_common",
  "aquatic_udp_protocol",
  "cfg-if",
+ "chrono",
  "crossbeam-channel",
  "hex",
  "log",
  "mimalloc",
  "mio",
+ "num-format",
  "parking_lot",
  "quickcheck",
  "quickcheck_macros",
@@ -176,6 +178,7 @@ dependencies = [
  "signal-hook",
  "slab",
  "socket2 0.4.2",
+ "tinytemplate",
 ]
 
 [[package]]

aquatic_udp/Cargo.toml

@@ -21,18 +21,21 @@ anyhow = "1"
 aquatic_cli_helpers = "0.1.0"
 aquatic_common = "0.1.0"
 aquatic_udp_protocol = "0.1.0"
+chrono = "0.4"
 cfg-if = "1"
 crossbeam-channel = "0.5"
 hex = "0.4"
 log = "0.4"
 mimalloc = { version = "0.1", default-features = false }
 mio = { version = "0.8", features = ["net", "os-poll"] }
+num-format = "0.4"
 parking_lot = "0.11"
 rand = { version = "0.8", features = ["small_rng"] }
 serde = { version = "1", features = ["derive"] }
 slab = "0.4"
 signal-hook = { version = "0.3" }
 socket2 = { version = "0.4", features = ["all"] }
+tinytemplate = "1"
 
 [dev-dependencies]
 quickcheck = "1"

aquatic_udp/src/config.rs

@@ -1,4 +1,4 @@
-use std::net::SocketAddr;
+use std::{net::SocketAddr, path::PathBuf};
 
 use aquatic_common::{access_list::AccessListConfig, privileges::PrivilegeConfig};
 use serde::{Deserialize, Serialize};
@@ -87,6 +87,15 @@ pub struct NetworkConfig {
     pub poll_timeout_ms: u64,
 }
 
+impl NetworkConfig {
+    pub fn ipv4_active(&self) -> bool {
+        self.address.is_ipv4() || !self.only_ipv6
+    }
+    pub fn ipv6_active(&self) -> bool {
+        self.address.is_ipv6()
+    }
+}
+
 impl Default for NetworkConfig {
     fn default() -> Self {
         Self {
@@ -123,13 +132,30 @@ impl Default for ProtocolConfig {
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(default)]
 pub struct StatisticsConfig {
-    /// Print statistics this often (seconds). Don't print when set to zero.
+    /// Collect and print/write statistics this often (seconds)
     pub interval: u64,
+    /// Print statistics to standard output
+    pub print_to_stdout: bool,
+    /// Save statistics as HTML to a file
+    pub write_html_to_file: bool,
+    /// Path to save HTML file
+    pub html_file_path: PathBuf,
+}
+
+impl StatisticsConfig {
+    pub fn active(&self) -> bool {
+        (self.interval != 0) & (self.print_to_stdout | self.write_html_to_file)
+    }
 }
 
 impl Default for StatisticsConfig {
     fn default() -> Self {
-        Self { interval: 0 }
+        Self {
+            interval: 5,
+            print_to_stdout: false,
+            write_html_to_file: false,
+            html_file_path: "tmp/statistics.html".into(),
+        }
     }
 }
 

aquatic_udp/src/lib.rs

@@ -118,7 +118,7 @@ pub fn run(config: Config) -> ::anyhow::Result<()> {
             .with_context(|| "spawn socket worker")?;
     }
 
-    if config.statistics.interval != 0 {
+    if config.statistics.active() {
         let state = state.clone();
         let config = config.clone();
 

aquatic_udp/src/workers/socket.rs

@@ -293,7 +293,7 @@ fn read_requests(
         }
     }
 
-    if config.statistics.interval != 0 {
+    if config.statistics.active() {
         state
             .statistics_ipv4
             .requests_received
@@ -471,7 +471,7 @@ fn send_responses(
         }
     }
 
-    if config.statistics.interval != 0 {
+    if config.statistics.active() {
         state
             .statistics_ipv4
             .responses_sent

aquatic_udp/src/workers/statistics.rs

@@ -1,12 +1,119 @@
+use std::fs::File;
+use std::io::Write;
 use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+use anyhow::Context;
+use chrono::Utc;
+use num_format::{Locale, ToFormattedString};
+use serde::Serialize;
+use tinytemplate::TinyTemplate;
+
 use crate::common::*;
 use crate::config::Config;
 
+const TEMPLATE_KEY: &str = "statistics";
+const TEMPLATE_CONTENTS: &str = include_str!("../../templates/statistics.html");
+const STYLESHEET_CONTENTS: &str = concat!(
+    "<style>",
+    include_str!("../../templates/statistics.css"),
+    "</style>"
+);
+
+#[derive(Clone, Copy, Debug)]
+struct CollectedStatistics {
+    requests_per_second: f64,
+    responses_per_second: f64,
+    bytes_received_per_second: f64,
+    bytes_sent_per_second: f64,
+    num_torrents: usize,
+    num_peers: usize,
+}
+
+impl CollectedStatistics {
+    fn from_shared(statistics: &Arc<Statistics>, last: &mut Instant) -> Self {
+        let requests_received = statistics.requests_received.fetch_and(0, Ordering::AcqRel) as f64;
+        let responses_sent = statistics.responses_sent.fetch_and(0, Ordering::AcqRel) as f64;
+        let bytes_received = statistics.bytes_received.fetch_and(0, Ordering::AcqRel) as f64;
+        let bytes_sent = statistics.bytes_sent.fetch_and(0, Ordering::AcqRel) as f64;
+        let num_torrents = Self::sum_atomic_usizes(&statistics.torrents);
+        let num_peers = Self::sum_atomic_usizes(&statistics.peers);
+
+        let now = Instant::now();
+
+        let elapsed = (now - *last).as_secs_f64();
+
+        *last = now;
+
+        Self {
+            requests_per_second: requests_received / elapsed,
+            responses_per_second: responses_sent / elapsed,
+            bytes_received_per_second: bytes_received / elapsed,
+            bytes_sent_per_second: bytes_sent / elapsed,
+            num_torrents,
+            num_peers,
+        }
+    }
+
+    fn sum_atomic_usizes(values: &[AtomicUsize]) -> usize {
+        values.iter().map(|n| n.load(Ordering::Acquire)).sum()
+    }
+}
+
+impl Into<FormattedStatistics> for CollectedStatistics {
+    fn into(self) -> FormattedStatistics {
+        let rx_mbits = self.bytes_received_per_second * 8.0 / 1_000_000.0;
+        let tx_mbits = self.bytes_sent_per_second * 8.0 / 1_000_000.0;
+
+        FormattedStatistics {
+            requests_per_second: (self.requests_per_second as usize)
+                .to_formatted_string(&Locale::en),
+            responses_per_second: (self.responses_per_second as usize)
+                .to_formatted_string(&Locale::en),
+            rx_mbits: format!("{:.2}", rx_mbits),
+            tx_mbits: format!("{:.2}", tx_mbits),
+            num_torrents: self.num_torrents.to_formatted_string(&Locale::en),
+            num_peers: self.num_peers.to_formatted_string(&Locale::en),
+        }
+    }
+}
+
+#[derive(Clone, Debug, Serialize)]
+struct FormattedStatistics {
+    requests_per_second: String,
+    responses_per_second: String,
+    rx_mbits: String,
+    tx_mbits: String,
+    num_torrents: String,
+    num_peers: String,
+}
+
+#[derive(Debug, Serialize)]
+struct TemplateData {
+    stylesheet: String,
+    ipv4_active: bool,
+    ipv6_active: bool,
+    ipv4: FormattedStatistics,
+    ipv6: FormattedStatistics,
+    last_updated: String,
+    peer_update_interval: String,
+}
+
 pub fn run_statistics_worker(config: Config, state: State) {
-    let ipv4_active = config.network.address.is_ipv4() || !config.network.only_ipv6;
-    let ipv6_active = config.network.address.is_ipv6();
+    let tt = if config.statistics.write_html_to_file {
+        let mut tt = TinyTemplate::new();
+
+        if let Err(err) = tt.add_template(TEMPLATE_KEY, TEMPLATE_CONTENTS) {
+            ::log::error!("Couldn't parse statistics html template: {:#}", err);
+
+            None
+        } else {
+            Some(tt)
+        }
+    } else {
+        None
+    };
 
     let mut last_ipv4 = Instant::now();
     let mut last_ipv6 = Instant::now();
@@ -14,60 +121,74 @@ pub fn run_statistics_worker(config: Config, state: State) {
     loop {
         ::std::thread::sleep(Duration::from_secs(config.statistics.interval));
 
-        println!("General:");
-        println!("  access list entries: {}", state.access_list.load().len());
+        let statistics_ipv4 =
+            CollectedStatistics::from_shared(&state.statistics_ipv4, &mut last_ipv4).into();
+        let statistics_ipv6 =
+            CollectedStatistics::from_shared(&state.statistics_ipv6, &mut last_ipv6).into();
 
-        if ipv4_active {
-            println!("IPv4:");
-            gather_and_print_for_protocol(&config, &state.statistics_ipv4, &mut last_ipv4);
-        }
-        if ipv6_active {
-            println!("IPv6:");
-            gather_and_print_for_protocol(&config, &state.statistics_ipv6, &mut last_ipv6);
+        if config.statistics.print_to_stdout {
+            println!("General:");
+            println!("  access list entries: {}", state.access_list.load().len());
+
+            if config.network.ipv4_active() {
+                println!("IPv4:");
+                print_to_stdout(&config, &statistics_ipv4);
+            }
+            if config.network.ipv6_active() {
+                println!("IPv6:");
+                print_to_stdout(&config, &statistics_ipv6);
+            }
+
+            println!();
         }
 
-        println!();
+        if let Some(tt) = tt.as_ref() {
+            let template_data = TemplateData {
+                stylesheet: STYLESHEET_CONTENTS.to_string(),
+                ipv4_active: config.network.ipv4_active(),
+                ipv6_active: config.network.ipv6_active(),
+                ipv4: statistics_ipv4,
+                ipv6: statistics_ipv6,
+                last_updated: Utc::now().to_rfc2822(),
+                peer_update_interval: format!("{}", config.cleaning.torrent_cleaning_interval),
+            };
+
+            if let Err(err) = save_html_to_file(&config, tt, &template_data) {
+                ::log::error!("Couldn't save statistics to file: {:#}", err)
+            }
+        }
     }
 }
 
-fn gather_and_print_for_protocol(config: &Config, statistics: &Statistics, last: &mut Instant) {
-    let requests_received: f64 = statistics.requests_received.fetch_and(0, Ordering::AcqRel) as f64;
-    let responses_sent: f64 = statistics.responses_sent.fetch_and(0, Ordering::AcqRel) as f64;
-    let bytes_received: f64 = statistics.bytes_received.fetch_and(0, Ordering::AcqRel) as f64;
-    let bytes_sent: f64 = statistics.bytes_sent.fetch_and(0, Ordering::AcqRel) as f64;
-
-    let now = Instant::now();
-
-    let elapsed = (now - *last).as_secs_f64();
-
-    *last = now;
-
-    let requests_per_second = requests_received / elapsed;
-    let responses_per_second: f64 = responses_sent / elapsed;
-    let bytes_received_per_second: f64 = bytes_received / elapsed;
-    let bytes_sent_per_second: f64 = bytes_sent / elapsed;
-
-    let num_torrents: usize = sum_atomic_usizes(&statistics.torrents);
-    let num_peers = sum_atomic_usizes(&statistics.peers);
-
+fn print_to_stdout(config: &Config, statistics: &FormattedStatistics) {
     println!(
-        "  requests/second: {:10.2}, responses/second: {:10.2}",
-        requests_per_second, responses_per_second
+        "  requests/second: {:>10}, responses/second: {:>10}",
+        statistics.requests_per_second, statistics.responses_per_second
     );
-
     println!(
-        "  bandwidth: {:7.2} Mbit/s in, {:7.2} Mbit/s out",
-        bytes_received_per_second * 8.0 / 1_000_000.0,
-        bytes_sent_per_second * 8.0 / 1_000_000.0,
+        "  bandwidth: {:>7} Mbit/s in, {:7} Mbit/s out",
+        statistics.rx_mbits, statistics.tx_mbits,
     );
-
-    println!("  number of torrents: {}", num_torrents);
+    println!("  number of torrents: {}", statistics.num_torrents);
     println!(
         "  number of peers: {} (updated every {} seconds)",
-        num_peers, config.cleaning.torrent_cleaning_interval
+        statistics.num_peers, config.cleaning.torrent_cleaning_interval
     );
 }
 
-fn sum_atomic_usizes(values: &[AtomicUsize]) -> usize {
-    values.iter().map(|n| n.load(Ordering::Acquire)).sum()
+fn save_html_to_file(
+    config: &Config,
+    tt: &TinyTemplate,
+    template_data: &TemplateData,
+) -> anyhow::Result<()> {
+    let mut file = File::create(&config.statistics.html_file_path).with_context(|| {
+        format!(
+            "File path: {}",
+            &config.statistics.html_file_path.to_string_lossy()
+        )
+    })?;
+
+    write!(file, "{}", tt.render(TEMPLATE_KEY, template_data)?)?;
+
+    Ok(())
 }

aquatic_udp/templates/statistics.css

@@ -0,0 +1,22 @@
+body {
+    font-family: arial, sans-serif;
+    font-size: 16px;
+}
+
+table {
+    border-collapse: collapse
+}
+
+caption {
+    caption-side: bottom;
+    padding-top: 0.5rem;
+}
+
+th, td {
+    padding: 0.5rem 2rem;
+    border: 1px solid #ccc;
+}
+
+th {
+    background-color: #eee;
+}

aquatic_udp/templates/statistics.html

@@ -0,0 +1,93 @@
+<!doctype html>
+
+<html lang="en">
+<head>
+    <meta charset="utf-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1">
+
+    <title>UDP BitTorrent tracker statistics</title>
+
+    {#- Include stylesheet like this to prevent code editor syntax warnings #}
+    { stylesheet | unescaped }
+</head>
+
+<body>
+    <h1>BitTorrent tracker statistics</h1>
+
+    <p>
+        <strong>Tracker software:</strong> <a href="https://github.com/greatest-ape/aquatic">aquatic_udp</a>
+    </p>
+
+    <p>
+        <strong>Updated:</strong> { last_updated } (UTC)
+    </p>
+
+    {{ if ipv4_active }}
+
+    <h2>IPv4</h2>
+
+    <table>
+        <caption>* Peer count is updated every { peer_update_interval } seconds</caption>
+        <tr>
+            <th scope="row">Number of torrents</th>
+            <td>{ ipv4.num_torrents }</td>
+        </tr>
+        <tr>
+            <th scope="row">Number of peers</th>
+            <td>{ ipv4.num_peers } *</td>
+        </tr>
+        <tr>
+            <th scope="row">Requests / second</th>
+            <td>{ ipv4.requests_per_second }</td>
+        </tr>
+        <tr>
+            <th scope="row">Responses / second</th>
+            <td>{ ipv4.responses_per_second }</td>
+        </tr>
+        <tr>
+            <th scope="row">Bandwidth (RX)</th>
+            <td>{ ipv4.rx_mbits } mbit/s</td>
+        </tr>
+        <tr>
+            <th scope="row">Bandwidth (TX)</th>
+            <td>{ ipv4.tx_mbits } mbit/s</td>
+        </tr>
+    </table>
+
+    {{ endif }}
+
+    {{ if ipv6_active }}
+
+    <h2>IPv6</h2>
+
+    <table>
+        <caption>* Peer count is updated every { peer_update_interval } seconds</caption>
+        <tr>
+            <th scope="row">Number of torrents</th>
+            <td>{ ipv6.num_torrents }</td>
+        </tr>
+        <tr>
+            <th scope="row">Number of peers</th>
+            <td>{ ipv6.num_peers } *</td>
+        </tr>
+        <tr>
+            <th scope="row">Requests / second</th>
+            <td>{ ipv6.requests_per_second }</td>
+        </tr>
+        <tr>
+            <th scope="row">Responses / second</th>
+            <td>{ ipv6.responses_per_second }</td>
+        </tr>
+        <tr>
+            <th scope="row">Bandwidth (RX)</th>
+            <td>{ ipv6.rx_mbits } mbit/s</td>
+        </tr>
+        <tr>
+            <th scope="row">Bandwidth (TX)</th>
+            <td>{ ipv6.tx_mbits } mbit/s</td>
+        </tr>
+    </table>
+
+    {{ endif }}
+</body>
+</html>