Merge pull request #16 from AstroHQ/feature/signpost-metric-sessions

Feature/signpost metric sessions & summary/average API

Author: jfro

.github/workflows/rust.yml

@@ -8,6 +8,7 @@ on:
 
 env:
   CARGO_TERM_COLOR: always
+  RUSTFLAGS: "-Dwarnings"
 
 jobs:
   build:
@@ -17,8 +18,12 @@ jobs:
     runs-on: ${{ matrix.os }}
 
     steps:
-    - uses: actions/checkout@v3
-    - name: Build
-      run: cargo build --verbose
-    - name: Run tests
-      run: cargo test --verbose
+      - uses: actions/checkout@v4
+      - name: Formatting
+        run: cargo fmt --check
+      - name: Build
+        run: cargo build --all --all-features
+      - name: Run tests
+        run: cargo test --verbose
+      - name: Run Clippy
+        run: cargo clippy --all-targets --all-features

Cargo.toml

@@ -12,19 +12,20 @@ documentation = "https://docs.rs/metrics-sqlite"
 repository = "https://github.com/AstroHQ/metrics-sqlite/"
 
 [dependencies]
-diesel = { version = "2.0.3", features = ["sqlite"] }
-diesel_migrations = "2.0.0"
+diesel = { version = "2.3", features = ["sqlite"] }
+diesel_migrations = "2.3"
 # this seems to let us force bundled sqlite in diesel
-libsqlite3-sys = { version = "0.31.0", features = ["bundled"] }
-metrics = "0.24.1"
-thiserror = "2.0.11"
+libsqlite3-sys = { version = "0.35.0", features = ["bundled"] }
+metrics = "0.24"
+thiserror = "2.0"
 tracing = "0.1"
 csv = { version = "1.1.6", optional = true }
-serde = { version = "1.0.125", optional = true }
+serde = { version = "1.0", optional = true }
+tokio = { version = "1.20", features = ["sync"] }
 
 [dev-dependencies]
-tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
-
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+anyhow = "1.0"
 
 [features]
 default = []

examples/summary.rs

@@ -0,0 +1,30 @@
+use metrics_sqlite::{MetricsDb, Session};
+use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
+
+fn main() -> anyhow::Result<()> {
+    let fmt_layer = fmt::layer();
+    let filter_layer = EnvFilter::try_from_default_env().or_else(|_| EnvFilter::try_new("info"))?;
+    tracing_subscriber::registry()
+        .with(filter_layer)
+        .with(fmt_layer)
+        .init();
+    let mut db = MetricsDb::new("metrics.db")?;
+    let session = db.session_from_signpost("liquid.connected")?;
+    let rtt = average(&mut db, "net_quality.rtt", &session)?;
+    let throughput = average(&mut db, "net_quality.throughput", &session)? / 1024.0; // KB/s
+    let metered_throughput =
+        average(&mut db, "rate_control.metered_throughput", &session)? / 1024.0; // KB/s
+    println!("Last connection: {session:#?}");
+    println!("RTT: Average {rtt:.2}ms");
+    println!("Throughput: Average {throughput:.2}KB/s");
+    println!("Metered: Average {metered_throughput:.2}KB/s");
+
+    Ok(())
+}
+fn average(db: &mut MetricsDb, key: &str, session: &Session) -> anyhow::Result<f64> {
+    let metrics = db.metrics_for_key(key, Some(session))?;
+    let sum: f64 = metrics.iter().map(|m| m.value).sum();
+    let samples = metrics.len();
+    let average = sum / samples as f64;
+    Ok(average)
+}

src/lib.rs

@@ -39,7 +39,7 @@ pub enum MetricsError {
     /// Error querying metrics DB
     #[error("Error querying DB: {0}")]
     QueryError(#[from] diesel::result::Error),
-    /// Error if path given is invalid
+    /// Error if the path given is invalid
     #[error("Invalid database path")]
     InvalidDatabasePath,
     /// IO Error with reader/writer
@@ -50,12 +50,21 @@ pub enum MetricsError {
     #[cfg(feature = "csv")]
     #[error("CSV Error: {0}")]
     CsvError(#[from] csv::Error),
-    /// Attempted to query database but found no records
+    /// Attempted to query the database but found no records
     #[error("Database has no metrics stored in it")]
     EmptyDatabase,
     /// Given metric key name wasn't found in the DB
     #[error("Metric key {0} not found in database")]
     KeyNotFound(String),
+    /// Attempting to communicate with exporter but it's gone away
+    #[error("Exporter task has been stopped or crashed")]
+    ExporterUnavailable,
+    /// Session derived from the signpost has zero duration
+    #[error("Session for signpost `{0}` has zero duration")]
+    ZeroLengthSession(String),
+    /// No metrics available for the requested key inside the derived session
+    #[error("No metrics recorded for `{0}` in requested session")]
+    NoMetricsForKey(String),
 }
 /// Metrics result type
 pub type Result<T, E = MetricsError> = std::result::Result<T, E>;
@@ -65,6 +74,7 @@ mod models;
 mod recorder;
 mod schema;
 
+use crate::metrics_db::query;
 use crate::recorder::Handle;
 pub use metrics_db::{MetricsDb, Session};
 pub use models::{Metric, MetricKey, NewMetric};
@@ -78,7 +88,7 @@ fn setup_db<P: AsRef<Path>>(path: P) -> Result<SqliteConnection> {
         .ok_or(MetricsError::InvalidDatabasePath)?;
     let mut db = SqliteConnection::establish(url)?;
     db.run_pending_migrations(MIGRATIONS)
-        .map_err(|e| MetricsError::MigrationError(e))?;
+        .map_err(MetricsError::MigrationError)?;
 
     Ok(db)
 }
@@ -101,9 +111,40 @@ enum Event {
         housekeeping_period: Option<Duration>,
         record_limit: Option<usize>,
     },
+    RequestSummaryFromSignpost {
+        signpost_key: String,
+        keys: Vec<String>,
+        tx: tokio::sync::oneshot::Sender<Result<HashMap<String, f64>>>,
+    },
+}
+
+/// Handle for continued communication with sqlite exporter
+pub struct SqliteExporterHandle {
+    sender: SyncSender<Event>,
+}
+impl SqliteExporterHandle {
+    /// Request average metrics from a signpost to latest from exporter's DB
+    pub fn request_average_metrics(
+        &self,
+        from_signpost: &str,
+        with_keys: &[&str],
+    ) -> Result<HashMap<String, f64>> {
+        let (tx, rx) = tokio::sync::oneshot::channel();
+        self.sender
+            .send(Event::RequestSummaryFromSignpost {
+                signpost_key: from_signpost.to_string(),
+                keys: with_keys.iter().map(|s| s.to_string()).collect(),
+                tx,
+            })
+            .map_err(|_| MetricsError::ExporterUnavailable)?;
+        match rx.blocking_recv() {
+            Ok(metrics) => Ok(metrics?),
+            Err(_) => Err(MetricsError::ExporterUnavailable),
+        }
+    }
 }
 
-/// Exports metrics by storing them in a SQLite database at a periodic interval
+/// Exports metrics by storing them in an SQLite database at a periodic interval
 pub struct SqliteExporter {
     thread: Option<JoinHandle<()>>,
     sender: SyncSender<Event>,
@@ -199,6 +240,16 @@ impl InnerState {
         self.queue.push_back(metric);
         Ok(())
     }
+
+    // --- Summary/Average additions
+
+    pub fn metrics_summary_for_signpost_and_keys(
+        &mut self,
+        signpost: String,
+        metrics: Vec<String>,
+    ) -> Result<HashMap<String, f64>> {
+        query::metrics_summary_for_signpost_and_keys(&mut self.db, &signpost, metrics)
+    }
 }
 
 fn run_worker(
@@ -292,6 +343,44 @@ fn run_worker(
 
                         (state.should_flush(), false)
                     }
+                    Ok(Event::RequestSummaryFromSignpost {
+                        signpost_key,
+                        keys,
+                        tx,
+                    }) => {
+                        match state.flush() {
+                            Ok(()) => match state
+                                .metrics_summary_for_signpost_and_keys(signpost_key, keys)
+                            {
+                                Ok(metrics) => {
+                                    if tx.send(Ok(metrics)).is_err() {
+                                        error!(
+                                            "Failed to respond with metrics results, discarding"
+                                        );
+                                    }
+                                }
+                                Err(e) => {
+                                    if let Err(e) = tx.send(Err(e)) {
+                                        error!(
+                                            "Failed to respond with metrics error result, discarding: {e:?}"
+                                        );
+                                    }
+                                }
+                            },
+                            Err(e) => {
+                                let err = MetricsError::from(e);
+                                error!(
+                                    "Failed to flush pending metrics before summary request: {err:?}"
+                                );
+                                if let Err(send_err) = tx.send(Err(err)) {
+                                    error!(
+                                        "Failed to respond with metrics flush error result, discarding: {send_err:?}"
+                                    );
+                                }
+                            }
+                        }
+                        (false, false)
+                    }
                     Err(RecvTimeoutError::Timeout) => {
                         debug!("Flushing due to {}s timeout", flush_duration.as_secs());
                         (true, false)
@@ -320,7 +409,7 @@ fn run_worker(
 }
 
 impl SqliteExporter {
-    /// Creates a new `SqliteExporter` that stores metrics in a SQLite database file.
+    /// Creates a new `SqliteExporter` that stores metrics in an SQLite database file.
     ///
     /// `flush_interval` specifies how often metrics are flushed to SQLite/disk
     ///
@@ -341,10 +430,10 @@ impl SqliteExporter {
         Ok(exporter)
     }
 
-    /// Sets optional periodic house keeping, None to disable (disabled by default)
+    /// Sets optional periodic housekeeping, None to disable (disabled by default)
     /// ## Notes
-    /// Periodic house keeping can affect metric recording, causing some data to be dropped during house keeping.
-    /// Record limit if set will cause anything over limit + 25% of limit to be removed
+    /// Periodic housekeeping can affect metric recording, causing some data to be dropped during housekeeping.
+    /// Record limit if set will cause anything over limit + 25% of the limit to be removed
     pub fn set_periodic_housekeeping(
         &self,
         periodic_duration: Option<Duration>,
@@ -409,7 +498,7 @@ impl SqliteExporter {
                             record_limit,
                             excess
                         );
-                        let query = format!("DELETE FROM metrics WHERE id IN (SELECT id FROM metrics ORDER BY timestamp ASC LIMIT {});", excess);
+                        let query = format!("DELETE FROM metrics WHERE id IN (SELECT id FROM metrics ORDER BY timestamp ASC LIMIT {excess});");
                         if let Err(e) = sql_query(query).execute(db) {
                             error!("Failed to delete excessive records: {:?}", e);
                         }
@@ -423,8 +512,12 @@ impl SqliteExporter {
     }
 
     /// Install recorder as `metrics` crate's Recorder
-    pub fn install(self) -> Result<(), SetRecorderError<Self>> {
-        metrics::set_global_recorder(self)
+    pub fn install(self) -> Result<SqliteExporterHandle, SetRecorderError<Self>> {
+        let handle = SqliteExporterHandle {
+            sender: self.sender.clone(),
+        };
+        metrics::set_global_recorder(self)?;
+        Ok(handle)
     }
 }
 impl Drop for SqliteExporter {