add metrics to collect date level stats of ingestion, query and other object store calls

Author: nikhilsinhaparseable

src/event/mod.rs

@@ -28,6 +28,7 @@ use self::error::EventError;
 use crate::{
     LOCK_EXPECT,
     metadata::update_stats,
+    metrics::{increment_events_ingested_by_date, increment_events_ingested_size_by_date},
     parseable::{PARSEABLE, StagingError},
     storage::StreamType,
 };
@@ -88,6 +89,11 @@ impl Event {
             self.parsed_timestamp.date(),
         );
 
+        // Track billing metrics for event ingestion
+        let date_string = self.parsed_timestamp.date().to_string();
+        increment_events_ingested_by_date(self.rb.num_rows() as u64, &date_string);
+        increment_events_ingested_size_by_date(self.origin_size, &date_string);
+
         crate::livetail::LIVETAIL.process(&self.stream_name, &self.rb);
 
         Ok(())

src/handlers/http/modal/ingest_server.rs

@@ -116,8 +116,6 @@ impl ParseableServer for IngestServer {
             })
             .await;
 
-        PARSEABLE.storage.register_store_metrics(prometheus);
-
         migration::run_migration(&PARSEABLE).await?;
 
         // local sync on init

src/handlers/http/modal/query_server.rs

@@ -106,7 +106,6 @@ impl ParseableServer for QueryServer {
         prometheus: &PrometheusMetrics,
         shutdown_rx: oneshot::Receiver<()>,
     ) -> anyhow::Result<()> {
-        PARSEABLE.storage.register_store_metrics(prometheus);
         // write the ingestor metadata to storage
         QUERIER_META
             .get_or_init(|| async {

src/handlers/http/modal/server.rs

@@ -128,8 +128,6 @@ impl ParseableServer for Server {
         prometheus: &PrometheusMetrics,
         shutdown_rx: oneshot::Receiver<()>,
     ) -> anyhow::Result<()> {
-        PARSEABLE.storage.register_store_metrics(prometheus);
-
         migration::run_migration(&PARSEABLE).await?;
 
         // load on init

src/handlers/http/query.rs

@@ -46,7 +46,7 @@ use tokio::task::JoinSet;
 use tracing::{error, warn};
 
 use crate::event::{DEFAULT_TIMESTAMP_KEY, commit_schema};
-use crate::metrics::QUERY_EXECUTE_TIME;
+use crate::metrics::{QUERY_EXECUTE_TIME, increment_query_calls_by_date};
 use crate::parseable::{PARSEABLE, StreamNotFound};
 use crate::query::error::ExecuteError;
 use crate::query::{CountsRequest, Query as LogicalQuery, execute};
@@ -123,6 +123,10 @@ pub async fn query(req: HttpRequest, query_request: Query) -> Result<HttpRespons
     user_auth_for_datasets(&permissions, &tables).await?;
     let time = Instant::now();
 
+    // Track billing metrics for query calls
+    let current_date = chrono::Utc::now().date_naive().to_string();
+    increment_query_calls_by_date(&current_date);
+
     // if the query is `select count(*) from <dataset>`
     // we use the `get_bin_density` method to get the count of records in the dataset
     // instead of executing the query using datafusion
@@ -341,6 +345,10 @@ pub async fn get_counts(
     req: HttpRequest,
     counts_request: Json<CountsRequest>,
 ) -> Result<impl Responder, QueryError> {
+    // Track billing metrics for query calls
+    let current_date = chrono::Utc::now().date_naive().to_string();
+    increment_query_calls_by_date(&current_date);
+
     let creds = extract_session_key_from_req(&req)?;
     let permissions = Users.get_permissions(&creds);
 

src/metrics/mod.rs

@@ -17,8 +17,6 @@
  */
 
 pub mod prom_utils;
-pub mod storage;
-
 use crate::{handlers::http::metrics_path, stats::FullStats};
 use actix_web::Responder;
 use actix_web_prometheus::{PrometheusMetrics, PrometheusMetricsBuilder};
@@ -228,6 +226,125 @@ pub static ALERTS_STATES: Lazy<IntCounterVec> = Lazy::new(|| {
     .expect("metric can be created")
 });
 
+// Billing Metrics - Counter type metrics for billing/usage tracking
+pub static TOTAL_EVENTS_INGESTED_BY_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "total_events_ingested_by_date",
+            "Total events ingested by date (Counter for billing)",
+        )
+        .namespace(METRICS_NAMESPACE),
+        &["date"],
+    )
+    .expect("metric can be created")
+});
+
+pub static TOTAL_EVENTS_INGESTED_SIZE_BY_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "total_events_ingested_size_by_date",
+            "Total events ingested size in bytes by date (Counter for billing)",
+        )
+        .namespace(METRICS_NAMESPACE),
+        &["date"],
+    )
+    .expect("metric can be created")
+});
+
+pub static TOTAL_PARQUETS_STORED_BY_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "total_parquets_stored_by_date",
+            "Total parquet files stored by date (Counter for billing)",
+        )
+        .namespace(METRICS_NAMESPACE),
+        &["date"],
+    )
+    .expect("metric can be created")
+});
+
+pub static TOTAL_PARQUETS_STORED_SIZE_BY_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "total_parquets_stored_size_by_date",
+            "Total parquet files stored size in bytes by date (Counter for billing)",
+        )
+        .namespace(METRICS_NAMESPACE),
+        &["date"],
+    )
+    .expect("metric can be created")
+});
+
+pub static TOTAL_QUERY_CALLS_BY_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "total_query_calls_by_date",
+            "Total query calls by date (Counter for billing)",
+        )
+        .namespace(METRICS_NAMESPACE),
+        &["date"],
+    )
+    .expect("metric can be created")
+});
+
+pub static TOTAL_FILES_SCANNED_IN_QUERY_BY_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "total_files_scanned_in_query_by_date",
+            "Total files scanned in queries by date (Counter for billing)",
+        )
+        .namespace(METRICS_NAMESPACE),
+        &["date"],
+    )
+    .expect("metric can be created")
+});
+
+pub static TOTAL_BYTES_SCANNED_IN_QUERY_BY_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "total_bytes_scanned_in_query_by_date",
+            "Total bytes scanned in queries by date (Counter for billing)",
+        )
+        .namespace(METRICS_NAMESPACE),
+        &["date"],
+    )
+    .expect("metric can be created")
+});
+
+pub static TOTAL_OBJECT_STORE_CALLS_BY_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
+    IntCounterVec::new(
+        Opts::new(
+            "total_object_store_calls_by_date",
+            "Total object store calls by date (Counter for billing)",
+        )
+        .namespace(METRICS_NAMESPACE),
+        &["provider", "method", "date"],
+    )
+    .expect("metric can be created")
+});
+
+pub static TOTAL_FILES_SCANNED_IN_OBJECT_STORE_CALLS_BY_DATE: Lazy<IntCounterVec> =
+    Lazy::new(|| {
+        IntCounterVec::new(
+            Opts::new(
+                "total_files_scanned_in_object_store_calls_by_date",
+                "Total files scanned in object store calls by date (Counter for billing)",
+            )
+            .namespace(METRICS_NAMESPACE),
+            &["provider", "method", "date"],
+        )
+        .expect("metric can be created")
+    });
+
+pub static STORAGE_REQUEST_RESPONSE_TIME: Lazy<HistogramVec> = Lazy::new(|| {
+    HistogramVec::new(
+        HistogramOpts::new("storage_request_response_time", "Storage Request Latency")
+            .namespace(METRICS_NAMESPACE),
+        &["provider", "method", "status"],
+    )
+    .expect("metric can be created")
+});
+
 fn custom_metrics(registry: &Registry) {
     registry
         .register(Box::new(EVENTS_INGESTED.clone()))
@@ -286,6 +403,39 @@ fn custom_metrics(registry: &Registry) {
     registry
         .register(Box::new(ALERTS_STATES.clone()))
         .expect("metric can be registered");
+    // Register billing metrics
+    registry
+        .register(Box::new(TOTAL_EVENTS_INGESTED_BY_DATE.clone()))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(TOTAL_EVENTS_INGESTED_SIZE_BY_DATE.clone()))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(TOTAL_PARQUETS_STORED_BY_DATE.clone()))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(TOTAL_PARQUETS_STORED_SIZE_BY_DATE.clone()))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(TOTAL_QUERY_CALLS_BY_DATE.clone()))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(TOTAL_FILES_SCANNED_IN_QUERY_BY_DATE.clone()))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(TOTAL_BYTES_SCANNED_IN_QUERY_BY_DATE.clone()))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(TOTAL_OBJECT_STORE_CALLS_BY_DATE.clone()))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(
+            TOTAL_FILES_SCANNED_IN_OBJECT_STORE_CALLS_BY_DATE.clone(),
+        ))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(STORAGE_REQUEST_RESPONSE_TIME.clone()))
+        .expect("metric can be registered");
 }
 
 pub fn build_metrics_handler() -> PrometheusMetrics {
@@ -345,6 +495,64 @@ pub async fn fetch_stats_from_storage(stream_name: &str, stats: FullStats) {
         .set(stats.lifetime_stats.storage as i64);
 }
 
+// Helper functions for tracking billing metrics
+pub fn increment_events_ingested_by_date(count: u64, date: &str) {
+    TOTAL_EVENTS_INGESTED_BY_DATE
+        .with_label_values(&[date])
+        .inc_by(count);
+}
+
+pub fn increment_events_ingested_size_by_date(size: u64, date: &str) {
+    TOTAL_EVENTS_INGESTED_SIZE_BY_DATE
+        .with_label_values(&[date])
+        .inc_by(size);
+}
+
+pub fn increment_parquets_stored_by_date(date: &str) {
+    TOTAL_PARQUETS_STORED_BY_DATE
+        .with_label_values(&[date])
+        .inc();
+}
+
+pub fn increment_parquets_stored_size_by_date(size: u64, date: &str) {
+    TOTAL_PARQUETS_STORED_SIZE_BY_DATE
+        .with_label_values(&[date])
+        .inc_by(size);
+}
+
+pub fn increment_query_calls_by_date(date: &str) {
+    TOTAL_QUERY_CALLS_BY_DATE.with_label_values(&[date]).inc();
+}
+
+pub fn increment_files_scanned_in_query_by_date(count: u64, date: &str) {
+    TOTAL_FILES_SCANNED_IN_QUERY_BY_DATE
+        .with_label_values(&[date])
+        .inc_by(count);
+}
+
+pub fn increment_bytes_scanned_in_query_by_date(bytes: u64, date: &str) {
+    TOTAL_BYTES_SCANNED_IN_QUERY_BY_DATE
+        .with_label_values(&[date])
+        .inc_by(bytes);
+}
+
+pub fn increment_object_store_calls_by_date(provider: &str, method: &str, date: &str) {
+    TOTAL_OBJECT_STORE_CALLS_BY_DATE
+        .with_label_values(&[provider, method, date])
+        .inc();
+}
+
+pub fn increment_files_scanned_in_object_store_calls_by_date(
+    provider: &str,
+    method: &str,
+    count: u64,
+    date: &str,
+) {
+    TOTAL_FILES_SCANNED_IN_OBJECT_STORE_CALLS_BY_DATE
+        .with_label_values(&[provider, method, date])
+        .inc_by(count);
+}
+
 use actix_web::HttpResponse;
 
 pub async fn get() -> Result<impl Responder, MetricsError> {