Add Prometheus metrics (#283)

This PR adds metrics for S3 / local store response times,
query performance, staging files, total events ingest 
and alerts.

Signed-off-by: Sebastian YEPES <[email protected]>
Co-authored-by: Satyam Singh <[email protected]>

Author: syepes

Cargo.lock

@@ -11,7 +11,7 @@ actix-web-httpauth = "0.8"
 actix-web = { version = "4.3", features = ["rustls"] }
 actix-cors = "0.6"
 actix-files = "0.6"
-actix-web-prometheus = { version = "0.1", features = ["process"] }
+actix-web-prometheus = { version = "0.1" }
 prometheus = { version = "0.13", features = ["process"] }
 anyhow = { version = "1.0", features = ["backtrace"] }
 arrow-schema = { version = "31.0", features = ["serde"] }
@@ -40,7 +40,7 @@ futures = "0.3"
 fs_extra = "1.3"
 http = "0.2"
 humantime-serde = "1.1"
-lazy_static = "1.4.0"
+lazy_static = "1.4"
 log = "0.4"
 num_cpus = "1.15"
 md-5 = "0.10"
@@ -67,9 +67,9 @@ actix-web-static-files = "4.0"
 static-files = "0.2"
 ulid = { version = "1.0", features = ["serde"] }
 ureq = { version = "2.6", features = ["json"] }
-hex = "0.4.3"
-itertools = "0.10.5"
-xxhash-rust = { version = "0.8.6", features = ["xxh3"] }
+hex = "0.4"
+itertools = "0.10"
+xxhash-rust = { version = "0.8", features = ["xxh3"] }
 
 [build-dependencies]
 static-files = "0.2"

server/Cargo.toml

@@ -17,10 +17,12 @@
  */
 
 use serde::{Deserialize, Serialize};
+use std::fmt;
 
 pub mod rule;
 pub mod target;
 
+use crate::metrics::ALERTS_STATES;
 use crate::utils::uid::Uid;
 
 pub use self::rule::Rule;
@@ -51,6 +53,13 @@ impl Alert {
             AlertState::Listening | AlertState::Firing => (),
             alert_state @ (AlertState::SetToFiring | AlertState::Resolved) => {
                 let context = self.get_context(stream_name, alert_state);
+                ALERTS_STATES
+                    .with_label_values(&[
+                        context.stream.as_str(),
+                        context.alert_name.as_str(),
+                        context.alert_state.to_string().as_str(),
+                    ])
+                    .inc();
                 for target in &self.targets {
                     target.call(context.clone());
                 }
@@ -124,3 +133,14 @@ impl Default for AlertState {
         Self::Listening
     }
 }
+
+impl fmt::Display for AlertState {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match *self {
+            AlertState::Listening => write!(f, "Listening"),
+            AlertState::SetToFiring => write!(f, "SetToFiring"),
+            AlertState::Firing => write!(f, "Firing"),
+            AlertState::Resolved => write!(f, "Resolved"),
+        }
+    }
+}

server/src/alerts/mod.rs

@@ -18,8 +18,10 @@
 
 use actix_web::{web, HttpRequest, HttpResponse};
 use serde_json::Value;
+use std::time::Instant;
 
 use crate::event;
+use crate::metrics::QUERY_EXECUTE_TIME;
 use crate::option::CONFIG;
 use crate::query::Query;
 use crate::response::QueryResponse;
@@ -34,17 +36,23 @@ const STREAM_NAME_HEADER_KEY: &str = "x-p-stream";
 const SEPARATOR: char = '^';
 
 pub async fn query(_req: HttpRequest, json: web::Json<Value>) -> Result<HttpResponse, QueryError> {
+    let time = Instant::now();
     let json = json.into_inner();
     let query = Query::parse(json)?;
 
     let storage = CONFIG.storage().get_object_store();
-
     let query_result = query.execute(storage).await;
-
-    query_result
+    let query_result = query_result
         .map(Into::<QueryResponse>::into)
         .map(|response| response.to_http())
-        .map_err(|e| e.into())
+        .map_err(|e| e.into());
+
+    let time = time.elapsed().as_secs_f64();
+    QUERY_EXECUTE_TIME
+        .with_label_values(&[query.stream_name.as_str()])
+        .observe(time);
+
+    query_result
 }
 
 // Handler for POST /api/v1/ingest

server/src/handlers/event.rs

@@ -21,7 +21,7 @@ use actix_web::dev::ServiceRequest;
 use actix_web::{middleware, web, App, HttpServer};
 use actix_web_httpauth::extractors::basic::BasicAuth;
 use actix_web_httpauth::middleware::HttpAuthentication;
-use actix_web_prometheus::{PrometheusMetrics, PrometheusMetricsBuilder};
+use actix_web_prometheus::PrometheusMetrics;
 use actix_web_static_files::ResourceFiles;
 use clokwerk::{AsyncScheduler, Scheduler, TimeUnits};
 use log::warn;
@@ -45,6 +45,7 @@ mod banner;
 mod event;
 mod handlers;
 mod metadata;
+mod metrics;
 mod migration;
 mod option;
 mod query;
@@ -69,11 +70,8 @@ async fn main() -> anyhow::Result<()> {
     CONFIG.validate_storage(&*storage).await;
     let metadata = storage::resolve_parseable_metadata().await?;
     banner::print(&CONFIG, metadata);
-    let prometheus = PrometheusMetricsBuilder::new(env!("CARGO_PKG_NAME"))
-        .registry(prometheus::default_registry().clone())
-        .endpoint("/metrics")
-        .build()
-        .unwrap();
+    let prometheus = metrics::build_metrics_handler();
+    CONFIG.storage().register_store_metrics(&prometheus);
 
     migration::run_migration(&CONFIG).await?;
 

server/src/main.rs

@@ -23,6 +23,7 @@ use std::sync::{Arc, RwLock};
 
 use crate::alerts::Alerts;
 use crate::event::Event;
+use crate::metrics::{EVENTS_INGESTED, EVENTS_INGESTED_SIZE};
 use crate::stats::{Stats, StatsCounter};
 use crate::storage::ObjectStorage;
 
@@ -180,6 +181,12 @@ impl STREAM_INFO {
 
         stream.stats.add_ingestion_size(size);
         stream.stats.increase_event_by_one();
+        EVENTS_INGESTED
+            .with_label_values(&[stream_name.clone(), "json"])
+            .inc();
+        EVENTS_INGESTED_SIZE
+            .with_label_values(&[stream_name.clone(), "json"])
+            .add(size as i64);
 
         Ok(())
     }