feat: Run queries via the ui for dashboard graphs

The prometheus library requires tokio.

Cargo.toml

@@ -10,12 +10,17 @@ license = "Apache-2.0"
 anyhow = "1.0.79"
 async-io = "2.3.1"
 axum = { version = "0.7.4", features = [ "ws" ] }
+axum-macros = "0.4.1"
 clap = { version = "4.4.18", features = ["derive"] }
 maud = { version = "0.26.0", features = ["axum"] }
-prometheus-http-api = "0.2.0"
+prometheus-http-query = "0.8.2"
 serde = { version = "1.0.196", features = ["derive"] }
 serde_json = "1.0.113"
 serde_yaml = "0.9.31"
 smol = "2.0.0"
 smol-axum = "0.1.0"
 smol-macros = "0.1.0"
+tokio = { version = "1.36.0", features = ["net", "rt", "rt-multi-thread"] }
+tower-http = { version = "0.5.1", features = ["trace"] }
+tracing = "0.1.40"
+tracing-subscriber = "0.3.18"

examples/example_dashboards.yaml

@@ -0,0 +1,17 @@
+---
+- title: Test Dasbboard 1
+  graphs:
+    - title: Node cpu
+      source: http://heimdall:9001
+      query: 'node_cpu_seconds_total{job="nodestats"}'
+    - title: Node memory
+      source: http://heimdall:9001
+      query: 'node_memory_MemFree_bytes{instance="andrew:9002",job="nodestats"}'
+- title: Test Dasbboard 2
+  graphs:
+    - title: Node cpu
+      source: http://heimdall:9001
+      query: 'node_cpu_seconds_total{job="nodestats"}'
+    - title: Node memory
+      source: http://heimdall:9001
+      query: 'node_memory_MemFree_bytes{instance="andrew:9002",job="nodestats"}'

src/dashboard.rs

@@ -15,6 +15,9 @@ use std::path::Path;
 
 use serde::Deserialize;
 use serde_yaml;
+use tracing::{debug, info};
+
+use crate::query::QueryConn;
 
 #[derive(Deserialize)]
 pub struct Dashboard {
@@ -25,9 +28,17 @@ pub struct Dashboard {
 #[derive(Deserialize)]
 pub struct Graph {
     pub title: String,
+    pub source: String,
     pub query: String,
 }
 
+impl Graph {
+    pub fn get_query_connection<'conn, 'graph: 'conn>(&'graph self) -> QueryConn<'conn> {
+        debug!(query=self.query, source=self.source, "Getting query connection for graph");
+        QueryConn::new(&self.source, &self.query)
+    }
+}
+
 pub fn read_dashboard_list(path: &Path) -> anyhow::Result<Vec<Dashboard>> {
     let f = std::fs::File::open(path)?;
     Ok(serde_yaml::from_reader(f)?)

src/main.rs

@@ -11,43 +11,66 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
-use std::net::TcpListener;
 use std::path::PathBuf;
-use std::sync::Arc;
 
 use anyhow;
-use async_io::Async;
 use axum::{self, extract::State, routing::*, Router};
-use clap::{self, Parser};
-use smol_macros::main;
+use clap::{self, Parser, ValueEnum};
+use tokio::net::TcpListener;
+use tower_http::trace::TraceLayer;
+use tracing::Level;
+use tracing_subscriber::FmtSubscriber;
 
 mod dashboard;
 mod query;
 mod routes;
 
+#[derive(Copy, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum)]
+enum Verbosity {
+    ERROR,
+    WARN,
+    INFO,
+    DEBUG,
+    TRACE,
+}
+
 #[derive(clap::Parser)]
 #[command(author, version, about, long_about = None)]
 struct Cli {
     #[arg(long)]
-    listen: Option<std::net::SocketAddr>,
+    pub listen: Option<String>,
     #[arg(long)]
-    config: PathBuf,
+    pub config: PathBuf,
+    #[arg(long, value_enum, default_value_t = Verbosity::INFO)]
+    pub verbose: Verbosity,
 }
 
-main! {
-    async fn main(ex: &Arc<smol_macros::Executor<'_>>) -> anyhow::Result<()> {
-        let args = Cli::parse();
-        let config = std::sync::Arc::new(dashboard::read_dashboard_list(args.config.as_path())?);
-        let router = Router::new()
-            // JSON api endpoints
-            .nest("/api", routes::mk_api_routes())
-            // HTMX ui component endpoints
-            .nest("/ui", routes::mk_ui_routes())
-            .route("/", get(routes::index).with_state(config.clone()))
-            .with_state(State(config.clone()));
-        let socket_addr = args.listen.unwrap_or("127.0.0.1:3000".parse()?);
-        let listener = Async::<TcpListener>::bind(socket_addr)?;
-        smol_axum::serve(ex.clone(), listener, router).await?;
-        Ok(())
-    }
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    let args = Cli::parse();
+    let subscriber_builder = FmtSubscriber::builder().with_max_level(match args.verbose {
+        Verbosity::ERROR => Level::ERROR,
+        Verbosity::WARN => Level::WARN,
+        Verbosity::INFO => Level::INFO,
+        Verbosity::DEBUG => Level::DEBUG,
+        Verbosity::TRACE => Level::TRACE,
+    });
+    tracing::subscriber::set_global_default(
+        subscriber_builder.with_writer(std::io::stderr).finish(),
+    )
+    .expect("setting default subscriber failed");
+
+    let config = std::sync::Arc::new(dashboard::read_dashboard_list(args.config.as_path())?);
+    let router = Router::new()
+        // JSON api endpoints
+        .nest("/api", routes::mk_api_routes(config.clone()))
+        // HTMX ui component endpoints
+        .nest("/ui", routes::mk_ui_routes(config.clone()))
+        .route("/", get(routes::index).with_state(config.clone()))
+        .layer(TraceLayer::new_for_http())
+        .with_state(State(config.clone()));
+    let socket_addr = args.listen.unwrap_or("127.0.0.1:3000".to_string());
+    let listener = TcpListener::bind(socket_addr).await.expect("Unable to bind listener to address");
+    axum::serve(listener, router).await?;
+    Ok(())
 }

src/query.rs

@@ -11,18 +11,63 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
-use prometheus_http_api::{DataSource, Query};
+use std::collections::HashMap;
 
-pub struct QueryConn {
-    source: DataSource,
-    query: Query,
+use prometheus_http_query::{Client, response::{PromqlResult, Data}};
+use serde::{Serialize, Deserialize};
+use tracing::debug;
+
+pub struct QueryConn<'conn> {
+    source: &'conn str,
+    query: &'conn str,
 }
 
-impl QueryConn {
-    pub fn new<S: Into<DataSource>, Q: Into<Query>>(src: S, qry: Q) -> Self {
+impl<'conn> QueryConn<'conn> {
+    pub fn new<'a: 'conn>(source: &'a str, query: &'a str) -> Self {
         Self {
-            source: src.into(),
-            query: qry.into(),
+            source,
+            query,
+        }
+    }
+
+    pub async fn get_results(&self) -> anyhow::Result<PromqlResult> {
+        debug!("Getting results for query");
+        let client = Client::try_from(self.source)?;
+        Ok(client.query(self.query).get().await?)
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct DataPoint {
+    timesstamp: f64,
+    value: f64,
+}
+
+
+#[derive(Serialize, Deserialize)]
+pub enum QueryResult {
+    Series(Vec<(HashMap<String, String>, Vec<DataPoint>)>),
+    Scalar(DataPoint),
+}
+
+pub fn to_samples(data: Data) -> QueryResult {
+    match data {
+        Data::Matrix(mut range) => {
+            QueryResult::Series(range.drain(0..).map(|rv| {
+                let (metric, mut samples) = rv.into_inner();
+                (metric, samples.drain(0..).map(|s| {
+                    DataPoint { timesstamp: s.timestamp(), value: s.value() }
+                }).collect())
+            }).collect())
+        }
+        Data::Vector(mut vector) => {
+            QueryResult::Series(vector.drain(0..).map(|iv| {
+                let (metric, sample) = iv.into_inner();
+                (metric, vec![DataPoint { timesstamp: sample.timestamp(), value: sample.value() }])
+            }).collect())
+        }
+        Data::Scalar(sample) => {
+            QueryResult::Scalar(DataPoint { timesstamp: sample.timestamp(), value: sample.value() })
         }
     }
 }

src/routes.rs

@@ -13,19 +13,52 @@
 // limitations under the License.
 use std::sync::Arc;
 
+use axum::{
+    extract::{Path, State},
+    routing::get,
+    Json, Router,
+};
 use maud::{html, Markup};
-use axum::{extract::State, Router};
+use tracing::debug;
 
 use crate::dashboard::Dashboard;
+use crate::query::{to_samples, QueryResult};
 
 type Config = State<Arc<Vec<Dashboard>>>;
 
-pub fn mk_api_routes() -> Router<Config> {
-    // Query routes
-    Router::new()
+//#[axum_macros::debug_handler]
+pub async fn graph_query(
+    State(config): Config,
+    Path((dash_idx, graph_idx)): Path<(usize, usize)>,
+) -> Json<QueryResult> {
+    debug!("Getting data for query");
+    let graph = config
+        .get(dash_idx)
+        .expect("No such dashboard index")
+        .graphs
+        .get(graph_idx)
+        .expect(&format!("No such graph in dasboard {}", dash_idx));
+    let data = to_samples(
+        graph
+            .get_query_connection()
+            .get_results()
+            .await
+            .expect("Unable to get query results")
+            .data()
+            .clone(),
+    );
+    Json(data)
 }
 
-pub fn mk_ui_routes() -> Router<Config> {
+pub fn mk_api_routes(config: Arc<Vec<Dashboard>>) -> Router<Config> {
+    // Query routes
+    Router::new().route(
+        "/dash/:dash_idx/graph/:graph_idx",
+        get(graph_query).with_state(config),
+    )
+}
+
+pub fn mk_ui_routes(config: Arc<Vec<Dashboard>>) -> Router<Config> {
     Router::new()
 }
 
@@ -43,7 +76,10 @@ pub async fn index(State(config): Config) -> Markup {
 }
 
 pub async fn app(State(config): Config) -> Markup {
-    let titles = config.iter().map(|d| d.title.clone()).collect::<Vec<String>>();
+    let titles = config
+        .iter()
+        .map(|d| d.title.clone())
+        .collect::<Vec<String>>();
     html! {
         div {
             // Header menu