fix: merge and DatedStats

Author: Devdutt Shenoi

src/analytics.rs

@@ -41,7 +41,7 @@ use crate::{
     },
     option::Mode,
     parseable::PARSEABLE,
-    stats::{self, Stats},
+    stats::{FullStats, Stats},
     storage, HTTP_CLIENT,
 };
 
@@ -170,7 +170,7 @@ fn total_event_stats() -> (Stats, Stats, Stats) {
     let mut deleted_json_bytes: u64 = 0;
 
     for stream in PARSEABLE.streams.list() {
-        let Some(stats) = stats::get_current_stats(&stream, "json") else {
+        let Some(stats) = FullStats::get_current(&stream, "json") else {
             continue;
         };
         total_events += stats.lifetime_stats.events;

src/catalog/mod.rs

@@ -33,7 +33,7 @@ use crate::{
     option::Mode,
     parseable::PARSEABLE,
     query::PartialTimeFilter,
-    stats::{event_labels_date, get_current_stats, storage_size_labels_date, update_deleted_stats},
+    stats::{event_labels_date, storage_size_labels_date, update_deleted_stats, FullStats},
     storage::{
         object_storage::manifest_path, ObjectStorage, ObjectStorageError, ObjectStoreFormat,
     },
@@ -181,7 +181,7 @@ pub async fn update_snapshot(
             if let Some(mut manifest) = storage.get_manifest(&path).await? {
                 manifest.apply_change(change);
                 storage.put_manifest(&path, manifest).await?;
-                let stats = get_current_stats(stream_name, "json");
+                let stats = FullStats::get_current(stream_name, "json");
                 if let Some(stats) = stats {
                     meta.stats = stats;
                 }
@@ -307,7 +307,7 @@ async fn create_manifest(
         };
         manifests.push(new_snapshot_entry);
         meta.snapshot.manifest_list = manifests;
-        let stats = get_current_stats(stream_name, "json");
+        let stats = FullStats::get_current(stream_name, "json");
         if let Some(stats) = stats {
             meta.stats = stats;
         }

src/handlers/http/cluster/mod.rs

@@ -25,7 +25,7 @@ use actix_web::http::header::{self, HeaderMap};
 use actix_web::web::Path;
 use actix_web::Responder;
 use bytes::Bytes;
-use chrono::Utc;
+use chrono::{NaiveDate, Utc};
 use clokwerk::{AsyncScheduler, Interval};
 use http::{header as http_header, StatusCode};
 use itertools::Itertools;
@@ -368,7 +368,7 @@ pub async fn sync_role_update_with_ingestors(
 }
 
 pub fn fetch_daily_stats_from_ingestors(
-    date: &str,
+    date: NaiveDate,
     stream_meta_list: &[ObjectStoreFormat],
 ) -> Result<Stats, StreamError> {
     // for the given date, get the stats from the ingestors
@@ -378,7 +378,7 @@ pub fn fetch_daily_stats_from_ingestors(
 
     for meta in stream_meta_list.iter() {
         for manifest in meta.snapshot.manifest_list.iter() {
-            if manifest.time_lower_bound.date_naive().to_string() == date {
+            if manifest.time_lower_bound.date_naive() == date {
                 events_ingested += manifest.events_ingested;
                 ingestion_size += manifest.ingestion_size;
                 storage_size += manifest.storage_size;

src/handlers/http/logstream.rs

@@ -35,7 +35,7 @@ use crate::metadata::SchemaVersion;
 use crate::parseable::{StreamNotFound, PARSEABLE};
 use crate::rbac::role::Action;
 use crate::rbac::Users;
-use crate::stats::StatsParams;
+use crate::stats::{FullStats, Stats, StatsParams};
 use crate::storage::retention::Retention;
 use crate::storage::{StreamInfo, StreamType};
 use crate::utils::actix::extract_session_key_from_req;
@@ -211,11 +211,12 @@ pub async fn get_stats(
         return Err(StreamNotFound(stream_name.clone()).into());
     }
 
-    if let Some(stats) = params.get_stats(&stream_name) {
+    if let Some(date) = params.date {
+        let stats = Stats::for_stream_on_date(date, &stream_name);
         return Ok(HttpResponse::build(StatusCode::OK).json(stats));
     }
 
-    let stats = stats::get_current_stats(&stream_name, "json")
+    let stats = FullStats::get_current(&stream_name, "json")
         .ok_or_else(|| StreamNotFound(stream_name.clone()))?;
     let time = Utc::now();
     let stats = {

src/prism/home/mod.rs

@@ -29,13 +29,10 @@ use tracing::error;
 use crate::{
     alerts::{get_alerts_info, AlertError, AlertsInfo, ALERTS},
     correlation::{CorrelationError, CORRELATIONS},
-    handlers::http::{
-        cluster::fetch_daily_stats_from_ingestors,
-        logstream::{error::StreamError, get_stats_date},
-    },
+    handlers::http::logstream::error::StreamError,
     parseable::PARSEABLE,
     rbac::{map::SessionKey, role::Action, Users},
-    stats::Stats,
+    stats::{DatedStats, Stats},
     storage::{ObjectStorageError, ObjectStoreFormat, STREAM_ROOT_DIRECTORY},
     users::{dashboards::DASHBOARDS, filters::FILTERS},
 };
@@ -47,14 +44,6 @@ struct StreamInfo {
     stats_summary: Stats,
 }
 
-#[derive(Debug, Serialize, Default)]
-struct DatedStats {
-    date: String,
-    events: u64,
-    ingestion_size: u64,
-    storage_size: u64,
-}
-
 #[derive(Debug, Serialize)]
 struct TitleAndId {
     title: String,
@@ -155,8 +144,8 @@ pub async fn generate_home_response(key: &SessionKey) -> Result<HomeResponse, Pr
                 .checked_sub_signed(chrono::Duration::days(i))
                 .ok_or_else(|| anyhow::Error::msg("Date conversion faield"))
                 .unwrap()
+                .date_naive()
         })
-        .map(|date| date.format("%Y-%m-%d").to_string())
         .collect_vec();
 
     let mut stream_details = Vec::new();
@@ -193,7 +182,11 @@ pub async fn generate_home_response(key: &SessionKey) -> Result<HomeResponse, Pr
     }
 
     for date in dates.into_iter() {
-        let dated_stats = stats_for_date(date, stream_wise_ingestor_stream_json.clone()).await?;
+        let Some(dated_stats) =
+            DatedStats::for_all_streams(date, &stream_wise_ingestor_stream_json).await?
+        else {
+            continue;
+        };
         summary.stats_summary.events += dated_stats.events;
         summary.stats_summary.ingestion += dated_stats.ingestion_size;
         summary.stats_summary.storage += dated_stats.storage_size;
@@ -213,28 +206,6 @@ pub async fn generate_home_response(key: &SessionKey) -> Result<HomeResponse, Pr
     })
 }
 
-async fn stats_for_date(
-    date: String,
-    stream_wise_meta: HashMap<String, Vec<ObjectStoreFormat>>,
-) -> Result<DatedStats, PrismHomeError> {
-    // collect stats for all the streams for the given date
-    let mut details = DatedStats {
-        date: date.clone(),
-        ..Default::default()
-    };
-
-    for (stream, meta) in stream_wise_meta {
-        let querier_stats = get_stats_date(&stream, &date).await?;
-        let ingestor_stats = fetch_daily_stats_from_ingestors(&date, &meta)?;
-        // collect date-wise stats for all streams
-        details.events += querier_stats.events + ingestor_stats.events;
-        details.ingestion_size += querier_stats.ingestion + ingestor_stats.ingestion;
-        details.storage_size += querier_stats.storage + ingestor_stats.storage;
-    }
-
-    Ok(details)
-}
-
 #[derive(Debug, thiserror::Error)]
 pub enum PrismHomeError {
     #[error("Error: {0}")]

src/prism/logstream/mod.rs

@@ -34,7 +34,7 @@ use crate::{
         query::update_schema_when_distributed,
     },
     parseable::{StreamNotFound, PARSEABLE},
-    stats,
+    stats::FullStats,
     storage::{retention::Retention, StreamInfo, StreamType},
     LOCK_EXPECT,
 };
@@ -93,7 +93,7 @@ async fn get_stream_schema_helper(stream_name: &str) -> Result<Arc<Schema>, Stre
 }
 
 async fn get_stats(stream_name: &str) -> Result<QueriedStats, PrismLogstreamError> {
-    let stats = stats::get_current_stats(stream_name, "json")
+    let stats = FullStats::get_current(stream_name, "json")
         .ok_or_else(|| StreamNotFound(stream_name.to_owned()))?;
 
     let ingestor_stats = if PARSEABLE