Merge branch 'main' into main

Signed-off-by: Devdutt Shenoi <[email protected]>

Author: Devdutt Shenoi

src/cli.rs

@@ -260,11 +260,12 @@ pub struct Options {
         help = "Set a fixed memory limit for query in GiB"
     )]
     pub query_memory_pool_size: Option<usize>,
-
+    // reduced the max row group size from 1048576
+    // smaller row groups help in faster query performance in multi threaded query
     #[arg(
         long,
         env = "P_PARQUET_ROW_GROUP_SIZE",
-        default_value = "1048576",
+        default_value = "262144",
         help = "Number of rows in a row group"
     )]
     pub row_group_size: usize,

src/handlers/airplane.rs

@@ -215,10 +215,14 @@ impl FlightService for AirServiceImpl {
             Status::permission_denied("User Does not have permission to access this")
         })?;
         let time = Instant::now();
-        let (records, _) = query
-            .execute(stream_name.clone())
-            .await
-            .map_err(|err| Status::internal(err.to_string()))?;
+
+        let stream_name_clone = stream_name.clone();
+        let (records, _) =
+            match tokio::task::spawn_blocking(move || query.execute(stream_name_clone)).await {
+                Ok(Ok((records, fields))) => (records, fields),
+                Ok(Err(e)) => return Err(Status::internal(e.to_string())),
+                Err(err) => return Err(Status::internal(err.to_string())),
+            };
 
         /*
         * INFO: No returning the schema with the data.

src/handlers/http/cluster/mod.rs

@@ -368,64 +368,29 @@ pub async fn sync_role_update_with_ingestors(
     Ok(())
 }
 
-pub async fn fetch_daily_stats_from_ingestors(
-    stream_name: &str,
+pub fn fetch_daily_stats_from_ingestors(
     date: &str,
+    stream_meta_list: &[ObjectStoreFormat],
 ) -> Result<Stats, StreamError> {
-    let mut total_events_ingested: u64 = 0;
-    let mut total_ingestion_size: u64 = 0;
-    let mut total_storage_size: u64 = 0;
-
-    let ingestor_infos = get_ingestor_info().await.map_err(|err| {
-        error!("Fatal: failed to get ingestor info: {:?}", err);
-        StreamError::Anyhow(err)
-    })?;
-    for ingestor in ingestor_infos.iter() {
-        let uri = Url::parse(&format!(
-            "{}{}/metrics",
-            &ingestor.domain_name,
-            base_path_without_preceding_slash()
-        ))
-        .map_err(|err| {
-            StreamError::Anyhow(anyhow::anyhow!("Invalid URL in Ingestor Metadata: {}", err))
-        })?;
-
-        let res = HTTP_CLIENT
-            .get(uri)
-            .header(header::AUTHORIZATION, &ingestor.token)
-            .header(header::CONTENT_TYPE, "application/json")
-            .send()
-            .await;
-
-        if let Ok(res) = res {
-            let text = res
-                .text()
-                .await
-                .map_err(|err| StreamError::Anyhow(anyhow::anyhow!("Request failed: {}", err)))?;
-            let lines: Vec<Result<String, std::io::Error>> =
-                text.lines().map(|line| Ok(line.to_owned())).collect_vec();
-
-            let sample = prometheus_parse::Scrape::parse(lines.into_iter())
-                .map_err(|err| {
-                    StreamError::Anyhow(anyhow::anyhow!(
-                        "Invalid URL in Ingestor Metadata: {}",
-                        err
-                    ))
-                })?
-                .samples;
-
-            let (events_ingested, ingestion_size, storage_size) =
-                Metrics::get_daily_stats_from_samples(sample, stream_name, date);
-            total_events_ingested += events_ingested;
-            total_ingestion_size += ingestion_size;
-            total_storage_size += storage_size;
+    // for the given date, get the stats from the ingestors
+    let mut events_ingested = 0;
+    let mut ingestion_size = 0;
+    let mut storage_size = 0;
+
+    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 {
+                events_ingested += manifest.events_ingested;
+                ingestion_size += manifest.ingestion_size;
+                storage_size += manifest.storage_size;
+            }
         }
     }
 
     let stats = Stats {
-        events: total_events_ingested,
-        ingestion: total_ingestion_size,
-        storage: total_storage_size,
+        events: events_ingested,
+        ingestion: ingestion_size,
+        storage: storage_size,
     };
     Ok(stats)
 }
@@ -475,17 +440,17 @@ pub async fn fetch_stats_from_ingestors(
         Utc::now(),
         IngestionStats::new(
             count,
-            format!("{} Bytes", ingestion_size),
+            ingestion_size,
             lifetime_count,
-            format!("{} Bytes", lifetime_ingestion_size),
+            lifetime_ingestion_size,
             deleted_count,
-            format!("{} Bytes", deleted_ingestion_size),
+            deleted_ingestion_size,
             "json",
         ),
         StorageStats::new(
-            format!("{} Bytes", storage_size),
-            format!("{} Bytes", lifetime_storage_size),
-            format!("{} Bytes", deleted_storage_size),
+            storage_size,
+            lifetime_storage_size,
+            deleted_storage_size,
             "parquet",
         ),
     );

src/handlers/http/cluster/utils.rs

@@ -93,22 +93,22 @@ impl ClusterInfo {
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct IngestionStats {
     pub count: u64,
-    pub size: String,
+    pub size: u64,
     pub format: String,
     pub lifetime_count: u64,
-    pub lifetime_size: String,
+    pub lifetime_size: u64,
     pub deleted_count: u64,
-    pub deleted_size: String,
+    pub deleted_size: u64,
 }
 
 impl IngestionStats {
     pub fn new(
         count: u64,
-        size: String,
+        size: u64,
         lifetime_count: u64,
-        lifetime_size: String,
+        lifetime_size: u64,
         deleted_count: u64,
-        deleted_size: String,
+        deleted_size: u64,
         format: &str,
     ) -> Self {
         Self {
@@ -125,14 +125,14 @@ impl IngestionStats {
 
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct StorageStats {
-    pub size: String,
+    pub size: u64,
     pub format: String,
-    pub lifetime_size: String,
-    pub deleted_size: String,
+    pub lifetime_size: u64,
+    pub deleted_size: u64,
 }
 
 impl StorageStats {
-    pub fn new(size: String, lifetime_size: String, deleted_size: String, format: &str) -> Self {
+    pub fn new(size: u64, lifetime_size: u64, deleted_size: u64, format: &str) -> Self {
         Self {
             size,
             format: format.to_string(),
@@ -155,71 +155,23 @@ pub fn merge_quried_stats(stats: Vec<QueriedStats>) -> QueriedStats {
             .fold(IngestionStats::default(), |acc, x| IngestionStats {
                 count: acc.count + x.count,
 
-                size: format!(
-                    "{} Bytes",
-                    acc.size.split(' ').collect_vec()[0]
-                        .parse::<u64>()
-                        .unwrap_or_default()
-                        + x.size.split(' ').collect_vec()[0]
-                            .parse::<u64>()
-                            .unwrap_or_default()
-                ),
+                size: acc.size + x.size,
                 format: x.format.clone(),
                 lifetime_count: acc.lifetime_count + x.lifetime_count,
-                lifetime_size: format!(
-                    "{} Bytes",
-                    acc.lifetime_size.split(' ').collect_vec()[0]
-                        .parse::<u64>()
-                        .unwrap_or_default()
-                        + x.lifetime_size.split(' ').collect_vec()[0]
-                            .parse::<u64>()
-                            .unwrap_or_default()
-                ),
+                lifetime_size: acc.lifetime_size + x.lifetime_size,
                 deleted_count: acc.deleted_count + x.deleted_count,
-                deleted_size: format!(
-                    "{} Bytes",
-                    acc.deleted_size.split(' ').collect_vec()[0]
-                        .parse::<u64>()
-                        .unwrap_or_default()
-                        + x.deleted_size.split(' ').collect_vec()[0]
-                            .parse::<u64>()
-                            .unwrap_or_default()
-                ),
+                deleted_size: acc.deleted_size + x.deleted_size,
             });
 
     let cumulative_storage =
         stats
             .iter()
             .map(|x| &x.storage)
             .fold(StorageStats::default(), |acc, x| StorageStats {
-                size: format!(
-                    "{} Bytes",
-                    acc.size.split(' ').collect_vec()[0]
-                        .parse::<u64>()
-                        .unwrap_or_default()
-                        + x.size.split(' ').collect_vec()[0]
-                            .parse::<u64>()
-                            .unwrap_or_default()
-                ),
+                size: acc.size + x.size,
                 format: x.format.clone(),
-                lifetime_size: format!(
-                    "{} Bytes",
-                    acc.lifetime_size.split(' ').collect_vec()[0]
-                        .parse::<u64>()
-                        .unwrap_or_default()
-                        + x.lifetime_size.split(' ').collect_vec()[0]
-                            .parse::<u64>()
-                            .unwrap_or_default()
-                ),
-                deleted_size: format!(
-                    "{} Bytes",
-                    acc.deleted_size.split(' ').collect_vec()[0]
-                        .parse::<u64>()
-                        .unwrap_or_default()
-                        + x.deleted_size.split(' ').collect_vec()[0]
-                            .parse::<u64>()
-                            .unwrap_or_default()
-                ),
+                lifetime_size: acc.lifetime_size + x.lifetime_size,
+                deleted_size: acc.deleted_size + x.deleted_size,
             });
 
     QueriedStats::new(

src/handlers/http/logstream.rs

@@ -268,17 +268,17 @@ pub async fn get_stats(
     let stats = {
         let ingestion_stats = IngestionStats::new(
             stats.current_stats.events,
-            format!("{} Bytes", stats.current_stats.ingestion),
+            stats.current_stats.ingestion,
             stats.lifetime_stats.events,
-            format!("{} Bytes", stats.lifetime_stats.ingestion),
+            stats.lifetime_stats.ingestion,
             stats.deleted_stats.events,
-            format!("{} Bytes", stats.deleted_stats.ingestion),
+            stats.deleted_stats.ingestion,
             "json",
         );
         let storage_stats = StorageStats::new(
-            format!("{} Bytes", stats.current_stats.storage),
-            format!("{} Bytes", stats.lifetime_stats.storage),
-            format!("{} Bytes", stats.deleted_stats.storage),
+            stats.current_stats.storage,
+            stats.lifetime_stats.storage,
+            stats.deleted_stats.storage,
             "parquet",
         );
 

src/handlers/http/modal/query/querier_logstream.rs

@@ -26,6 +26,7 @@ use actix_web::{
 use bytes::Bytes;
 use chrono::Utc;
 use http::StatusCode;
+use relative_path::RelativePathBuf;
 use tokio::sync::Mutex;
 use tracing::{error, warn};
 
@@ -44,7 +45,7 @@ use crate::{
     hottier::HotTierManager,
     parseable::{StreamNotFound, PARSEABLE},
     stats::{self, Stats},
-    storage::StreamType,
+    storage::{ObjectStoreFormat, StreamType, STREAM_ROOT_DIRECTORY},
 };
 
 pub async fn delete(stream_name: Path<String>) -> Result<impl Responder, StreamError> {
@@ -151,7 +152,35 @@ pub async fn get_stats(
 
         if !date_value.is_empty() {
             let querier_stats = get_stats_date(&stream_name, date_value).await?;
-            let ingestor_stats = fetch_daily_stats_from_ingestors(&stream_name, date_value).await?;
+
+            // this function requires all the ingestor stream jsons
+            let path = RelativePathBuf::from_iter([&stream_name, STREAM_ROOT_DIRECTORY]);
+            let obs = PARSEABLE
+                .storage
+                .get_object_store()
+                .get_objects(
+                    Some(&path),
+                    Box::new(|file_name| {
+                        file_name.starts_with(".ingestor") && file_name.ends_with("stream.json")
+                    }),
+                )
+                .await?;
+
+            let mut ingestor_stream_jsons = Vec::new();
+            for ob in obs {
+                let stream_metadata: ObjectStoreFormat = match serde_json::from_slice(&ob) {
+                    Ok(d) => d,
+                    Err(e) => {
+                        error!("Failed to parse stream metadata: {:?}", e);
+                        continue;
+                    }
+                };
+                ingestor_stream_jsons.push(stream_metadata);
+            }
+
+            let ingestor_stats =
+                fetch_daily_stats_from_ingestors(date_value, &ingestor_stream_jsons)?;
+
             let total_stats = Stats {
                 events: querier_stats.events + ingestor_stats.events,
                 ingestion: querier_stats.ingestion + ingestor_stats.ingestion,
@@ -180,17 +209,17 @@ pub async fn get_stats(
     let stats = {
         let ingestion_stats = IngestionStats::new(
             stats.current_stats.events,
-            format!("{} Bytes", stats.current_stats.ingestion),
+            stats.current_stats.ingestion,
             stats.lifetime_stats.events,
-            format!("{} Bytes", stats.lifetime_stats.ingestion),
+            stats.lifetime_stats.ingestion,
             stats.deleted_stats.events,
-            format!("{} Bytes", stats.deleted_stats.ingestion),
+            stats.deleted_stats.ingestion,
             "json",
         );
         let storage_stats = StorageStats::new(
-            format!("{} Bytes", stats.current_stats.storage),
-            format!("{} Bytes", stats.lifetime_stats.storage),
-            format!("{} Bytes", stats.deleted_stats.storage),
+            stats.current_stats.storage,
+            stats.lifetime_stats.storage,
+            stats.deleted_stats.storage,
             "parquet",
         );