Draft: Prometheus from control plane

crates/snot-agent/src/rpc.rs

@@ -427,6 +427,32 @@ impl AgentService for AgentRpcServer {
         }
     }
 
+    async fn get_metrics(self, _: context::Context) -> Result<String, AgentError> {
+        if !matches!(
+            self.state.agent_state.read().await.deref(),
+            AgentState::Node(_, _)
+        ) {
+            return Err(AgentError::InvalidState);
+        }
+
+        let url = format!("http://127.0.0.1:{}/", self.state.cli.ports.metrics);
+
+        let response = reqwest::Client::new()
+            .get(url)
+            .send()
+            .await
+            .map_err(|_| AgentError::FailedToMakeRequest)?;
+
+        if !response.status().is_success() {
+            return Err(AgentError::FailedToMakeRequest);
+        }
+
+        response
+            .text()
+            .await
+            .map_err(|_| AgentError::FailedToMakeRequest)
+    }
+
     async fn get_metric(self, _: context::Context, metric: AgentMetric) -> f64 {
         let metrics = self.state.metrics.read().await;
 

crates/snot-common/src/rpc/agent.rs

@@ -27,13 +27,16 @@ pub trait AgentService {
     async fn broadcast_tx(tx: String) -> Result<(), AgentError>;
 
     /// Locally execute an authorization, using the given query
-    /// environment id is passed so the agent can determine which aot binary to use
+    /// environment id is passed so the agent can determine which aot binary to
+    /// use
     async fn execute_authorization(
         env_id: usize,
         query: String,
         auth: String,
     ) -> Result<(), AgentError>;
 
+    async fn get_metrics() -> Result<String, AgentError>;
+
     async fn get_metric(metric: AgentMetric) -> f64;
 }
 

crates/snot-common/src/rpc/control.rs

@@ -10,8 +10,6 @@ use crate::state::AgentId;
 
 #[tarpc::service]
 pub trait ControlService {
-    async fn placeholder() -> String;
-
     /// Resolve the addresses of the given agents.
     async fn resolve_addrs(
         peers: HashSet<AgentId>,

crates/snot/Cargo.toml

@@ -40,3 +40,4 @@ tracing-subscriber.workspace = true
 url = { workspace = true, features = ["serde"] }
 uuid = { workspace = true, features = ["v4", "fast-rng"] }
 wildmatch = "2.3.3"
+bollard = "0.16.1"

crates/snot/src/cli.rs

@@ -8,6 +8,10 @@ pub struct Cli {
     /// Control plane server port
     pub port: u16,
 
+    #[arg(long, default_value = "9090")]
+    /// Prometheus server port
+    pub prometheus: u16,
+
     #[arg(long, default_value = "snot-control-data")]
     /// Path to the directory containing the stored data
     pub path: PathBuf,

crates/snot/src/main.rs

@@ -1,14 +1,18 @@
-use std::io;
+use std::{io, sync::Arc};
 
 use clap::Parser;
 use cli::Cli;
+use surrealdb::Surreal;
 use tracing::level_filters::LevelFilter;
 use tracing_subscriber::prelude::*;
 
+use crate::state::GlobalState;
+
 pub mod cannon;
 pub mod cli;
 pub mod env;
 pub mod logging;
+pub mod prometheus;
 pub mod schema;
 pub mod server;
 pub mod state;
@@ -31,6 +35,8 @@ async fn main() {
         .add_directive("surrealdb=off".parse().unwrap())
         .add_directive("tungstenite=off".parse().unwrap())
         .add_directive("tokio_tungstenite=off".parse().unwrap())
+        .add_directive("tokio_util=off".parse().unwrap())
+        .add_directive("bollard=ERROR".parse().unwrap())
         .add_directive("tarpc::client=ERROR".parse().unwrap())
         .add_directive("tarpc::server=ERROR".parse().unwrap())
         .add_directive("tower_http::trace::on_request=off".parse().unwrap())
@@ -56,5 +62,26 @@ async fn main() {
 
     let cli = Cli::parse();
 
-    server::start(cli).await.expect("start server");
+    let mut path = cli.path.clone();
+    path.push("data.db");
+
+    let db = Surreal::new::<surrealdb::engine::local::File>(path)
+        .await
+        .expect("failed to create surrealDB");
+
+    let state = GlobalState {
+        cli,
+        db,
+        prom_ctr: Default::default(),
+        pool: Default::default(),
+        storage_ids: Default::default(),
+        storage: Default::default(),
+        envs: Default::default(),
+    };
+
+    prometheus::init(&state)
+        .await
+        .expect("failed to launch prometheus container");
+
+    server::start(Arc::new(state)).await.expect("start server");
 }

crates/snot/src/prometheus/config.rs

@@ -0,0 +1,47 @@
+use indexmap::IndexMap;
+use serde::Serialize;
+
+// TODO: we could probably clean this up or make it look a little bit prettier
+// later
+
+#[derive(Debug, Clone, Serialize)]
+pub struct PrometheusConfig {
+    pub global: GlobalConfig,
+    pub scrape_configs: Vec<ScrapeConfig>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct GlobalConfig {
+    pub scrape_interval: String,
+    pub scrape_timeout: String,
+    pub evaluation_interval: String,
+}
+
+impl Default for GlobalConfig {
+    fn default() -> Self {
+        Self {
+            scrape_interval: "15s".into(),
+            scrape_timeout: "10s".into(),
+            evaluation_interval: "1m".into(),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct ScrapeConfig {
+    pub job_name: String,
+    pub honor_timestamps: Option<bool>,
+    pub scrape_interval: Option<String>,
+    pub scrape_timeout: Option<String>,
+    pub metrics_path: Option<String>,
+    pub scheme: Option<String>,
+    pub follow_redirects: Option<bool>,
+    #[serde(default)]
+    pub static_configs: Vec<StaticConfig>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct StaticConfig {
+    pub targets: Vec<String>,
+    pub labels: IndexMap<String, String>,
+}