Merge remote-tracking branch 'origin/main'

Author: Devdutt Shenoi

src/alerts/mod.rs

@@ -24,6 +24,7 @@ use datafusion::common::tree_node::TreeNode;
 use http::StatusCode;
 use itertools::Itertools;
 use once_cell::sync::Lazy;
+use serde::Serialize;
 use serde_json::Error as SerdeError;
 use std::collections::{HashMap, HashSet};
 use std::fmt::{self, Display};
@@ -873,3 +874,52 @@ impl Alerts {
         Ok(())
     }
 }
+
+#[derive(Debug, Serialize)]
+pub struct AlertsInfo {
+    total: u64,
+    silenced: u64,
+    resolved: u64,
+    triggered: u64,
+    low: u64,
+    medium: u64,
+    high: u64,
+}
+
+// TODO: add RBAC
+pub async fn get_alerts_info() -> Result<AlertsInfo, AlertError> {
+    let alerts = ALERTS.alerts.read().await;
+    let mut total = 0;
+    let mut silenced = 0;
+    let mut resolved = 0;
+    let mut triggered = 0;
+    let mut low = 0;
+    let mut medium = 0;
+    let mut high = 0;
+
+    for (_, alert) in alerts.iter() {
+        total += 1;
+        match alert.state {
+            AlertState::Silenced => silenced += 1,
+            AlertState::Resolved => resolved += 1,
+            AlertState::Triggered => triggered += 1,
+        }
+
+        match alert.severity {
+            Severity::Low => low += 1,
+            Severity::Medium => medium += 1,
+            Severity::High => high += 1,
+            _ => {}
+        }
+    }
+
+    Ok(AlertsInfo {
+        total,
+        silenced,
+        resolved,
+        triggered,
+        low,
+        medium,
+        high,
+    })
+}

src/catalog/mod.rs

@@ -340,6 +340,12 @@ pub async fn remove_manifest_from_snapshot(
             Ok(get_first_event(storage.clone(), stream_name, Vec::new()).await?)
         }
         Mode::Query => Ok(get_first_event(storage, stream_name, dates).await?),
+        Mode::Index => Err(ObjectStorageError::UnhandledError(Box::new(
+            std::io::Error::new(
+                std::io::ErrorKind::Unsupported,
+                "Can't remove manifest from within Index server",
+            ),
+        ))),
     }
 }
 
@@ -350,6 +356,7 @@ pub async fn get_first_event(
 ) -> Result<Option<String>, ObjectStorageError> {
     let mut first_event_at: String = String::default();
     match PARSEABLE.options.mode {
+        Mode::Index => unimplemented!(),
         Mode::All | Mode::Ingest => {
             // get current snapshot
             let stream_first_event = PARSEABLE.get_stream(stream_name)?.get_first_event();

src/enterprise/mod.rs

@@ -0,0 +1 @@
+pub mod utils;

src/enterprise/utils.rs

@@ -0,0 +1,159 @@
+use std::{collections::HashMap, path::PathBuf, sync::Arc};
+
+use datafusion::{common::Column, prelude::Expr};
+use itertools::Itertools;
+use relative_path::RelativePathBuf;
+
+use crate::query::stream_schema_provider::extract_primary_filter;
+use crate::{
+    catalog::{
+        manifest::{File, Manifest},
+        snapshot, Snapshot,
+    },
+    event,
+    parseable::PARSEABLE,
+    query::{stream_schema_provider::ManifestExt, PartialTimeFilter},
+    storage::{ObjectStorage, ObjectStorageError, ObjectStoreFormat, STREAM_ROOT_DIRECTORY},
+    utils::time::TimeRange,
+};
+
+pub fn create_time_filter(
+    time_range: &TimeRange,
+    time_partition: Option<String>,
+    table_name: &str,
+) -> Vec<Expr> {
+    let mut new_filters = vec![];
+    let start_time = time_range.start.naive_utc();
+    let end_time = time_range.end.naive_utc();
+    let mut _start_time_filter: Expr;
+    let mut _end_time_filter: Expr;
+
+    match time_partition {
+        Some(time_partition) => {
+            _start_time_filter = PartialTimeFilter::Low(std::ops::Bound::Included(start_time))
+                .binary_expr(Expr::Column(Column::new(
+                    Some(table_name.to_owned()),
+                    time_partition.clone(),
+                )));
+            _end_time_filter =
+                PartialTimeFilter::High(std::ops::Bound::Excluded(end_time)).binary_expr(
+                    Expr::Column(Column::new(Some(table_name.to_owned()), time_partition)),
+                );
+        }
+        None => {
+            _start_time_filter = PartialTimeFilter::Low(std::ops::Bound::Included(start_time))
+                .binary_expr(Expr::Column(Column::new(
+                    Some(table_name.to_owned()),
+                    event::DEFAULT_TIMESTAMP_KEY,
+                )));
+            _end_time_filter = PartialTimeFilter::High(std::ops::Bound::Excluded(end_time))
+                .binary_expr(Expr::Column(Column::new(
+                    Some(table_name.to_owned()),
+                    event::DEFAULT_TIMESTAMP_KEY,
+                )));
+        }
+    }
+
+    new_filters.push(_start_time_filter);
+    new_filters.push(_end_time_filter);
+
+    new_filters
+}
+
+pub async fn fetch_parquet_file_paths(
+    stream: &str,
+    time_range: &TimeRange,
+) -> Result<HashMap<RelativePathBuf, Vec<File>>, ObjectStorageError> {
+    let glob_storage = PARSEABLE.storage.get_object_store();
+
+    let object_store_format = glob_storage.get_object_store_format(stream).await?;
+
+    let time_partition = object_store_format.time_partition;
+
+    let time_filter_expr = create_time_filter(time_range, time_partition.clone(), stream);
+
+    let time_filters = extract_primary_filter(&time_filter_expr, &time_partition);
+
+    let mut merged_snapshot: snapshot::Snapshot = snapshot::Snapshot::default();
+
+    let path = RelativePathBuf::from_iter([stream, STREAM_ROOT_DIRECTORY]);
+    let obs = glob_storage
+        .get_objects(
+            Some(&path),
+            Box::new(|file_name| file_name.ends_with("stream.json")),
+        )
+        .await;
+    if let Ok(obs) = obs {
+        for ob in obs {
+            if let Ok(object_store_format) = serde_json::from_slice::<ObjectStoreFormat>(&ob) {
+                let snapshot = object_store_format.snapshot;
+                for manifest in snapshot.manifest_list {
+                    merged_snapshot.manifest_list.push(manifest);
+                }
+            }
+        }
+    }
+
+    let manifest_files = collect_manifest_files(
+        glob_storage,
+        merged_snapshot
+            .manifests(&time_filters)
+            .into_iter()
+            .sorted_by_key(|file| file.time_lower_bound)
+            .map(|item| item.manifest_path)
+            .collect(),
+    )
+    .await?;
+
+    let mut parquet_files: HashMap<RelativePathBuf, Vec<File>> = HashMap::new();
+
+    let mut selected_files = manifest_files
+        .into_iter()
+        .flat_map(|file| file.files)
+        .rev()
+        .collect_vec();
+
+    for filter in time_filter_expr {
+        selected_files.retain(|file| !file.can_be_pruned(&filter))
+    }
+
+    selected_files
+        .into_iter()
+        .map(|file| {
+            let date = file.file_path.split("/").collect_vec();
+
+            let date = date.as_slice()[1..4].iter().map(|s| s.to_string());
+
+            let date = RelativePathBuf::from_iter(date);
+
+            parquet_files.entry(date).or_default().push(file);
+        })
+        .for_each(|_| {});
+
+    Ok(parquet_files)
+}
+
+async fn collect_manifest_files(
+    storage: Arc<dyn ObjectStorage>,
+    manifest_urls: Vec<String>,
+) -> Result<Vec<Manifest>, ObjectStorageError> {
+    let mut tasks = Vec::new();
+    manifest_urls.into_iter().for_each(|path| {
+        let path = RelativePathBuf::from_path(PathBuf::from(path)).expect("Invalid path");
+        let storage = Arc::clone(&storage);
+        tasks.push(tokio::task::spawn(async move {
+            storage.get_object(&path).await
+        }));
+    });
+
+    let mut op = Vec::new();
+    for task in tasks {
+        let file = task.await??;
+        op.push(file);
+    }
+
+    Ok(op
+        .into_iter()
+        .map(|res| serde_json::from_slice(&res).expect("Data is invalid for Manifest"))
+        .collect())
+}

src/handlers/http/logstream.rs

@@ -17,7 +17,7 @@
  */
 
 use self::error::StreamError;
-use super::cluster::utils::{merge_quried_stats, IngestionStats, QueriedStats, StorageStats};
+use super::cluster::utils::{IngestionStats, QueriedStats, StorageStats};
 use super::query::update_schema_when_distributed;
 use crate::event::format::override_data_type;
 use crate::hottier::{HotTierManager, StreamHotTier, CURRENT_HOT_TIER_VERSION};
@@ -263,64 +263,26 @@ pub async fn get_stats(
     let stats = stats::get_current_stats(&stream_name, "json")
         .ok_or_else(|| StreamNotFound(stream_name.clone()))?;
 
-    let ingestor_stats: Option<Vec<QueriedStats>> = None;
-
-    let hash_map = PARSEABLE.streams.read().expect("Readable");
-    let stream_meta = &hash_map
-        .get(&stream_name)
-        .ok_or_else(|| StreamNotFound(stream_name.clone()))?
-        .metadata
-        .read()
-        .expect(LOCK_EXPECT);
-
     let time = Utc::now();
 
-    let stats = match &stream_meta.first_event_at {
-        Some(_) => {
-            let ingestion_stats = IngestionStats::new(
-                stats.current_stats.events,
-                format!("{} {}", stats.current_stats.ingestion, "Bytes"),
-                stats.lifetime_stats.events,
-                format!("{} {}", stats.lifetime_stats.ingestion, "Bytes"),
-                stats.deleted_stats.events,
-                format!("{} {}", stats.deleted_stats.ingestion, "Bytes"),
-                "json",
-            );
-            let storage_stats = StorageStats::new(
-                format!("{} {}", stats.current_stats.storage, "Bytes"),
-                format!("{} {}", stats.lifetime_stats.storage, "Bytes"),
-                format!("{} {}", stats.deleted_stats.storage, "Bytes"),
-                "parquet",
-            );
-
-            QueriedStats::new(&stream_name, time, ingestion_stats, storage_stats)
-        }
-
-        None => {
-            let ingestion_stats = IngestionStats::new(
-                stats.current_stats.events,
-                format!("{} {}", stats.current_stats.ingestion, "Bytes"),
-                stats.lifetime_stats.events,
-                format!("{} {}", stats.lifetime_stats.ingestion, "Bytes"),
-                stats.deleted_stats.events,
-                format!("{} {}", stats.deleted_stats.ingestion, "Bytes"),
-                "json",
-            );
-            let storage_stats = StorageStats::new(
-                format!("{} {}", stats.current_stats.storage, "Bytes"),
-                format!("{} {}", stats.lifetime_stats.storage, "Bytes"),
-                format!("{} {}", stats.deleted_stats.storage, "Bytes"),
-                "parquet",
-            );
-
-            QueriedStats::new(&stream_name, time, ingestion_stats, storage_stats)
-        }
-    };
-    let stats = if let Some(mut ingestor_stats) = ingestor_stats {
-        ingestor_stats.push(stats);
-        merge_quried_stats(ingestor_stats)
-    } else {
-        stats
+    let stats = {
+        let ingestion_stats = IngestionStats::new(
+            stats.current_stats.events,
+            format!("{} Bytes", stats.current_stats.ingestion),
+            stats.lifetime_stats.events,
+            format!("{} Bytes", stats.lifetime_stats.ingestion),
+            stats.deleted_stats.events,
+            format!("{} Bytes", 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),
+            "parquet",
+        );
+
+        QueriedStats::new(&stream_name, time, ingestion_stats, storage_stats)
     };
 
     let stats = serde_json::to_value(stats)?;

src/handlers/http/middleware.rs

@@ -357,6 +357,25 @@ where
                     Ok(res)
                 })
             }
+
+            Mode::Index => {
+                let accessable_endpoints = ["create", "delete"];
+                let cond = path.split('/').any(|x| accessable_endpoints.contains(&x));
+                if !cond {
+                    Box::pin(async {
+                        Err(actix_web::error::ErrorUnauthorized(
+                            "Only Index API can be accessed in Index Mode",
+                        ))
+                    })
+                } else {
+                    let fut = self.service.call(req);
+
+                    Box::pin(async move {
+                        let res = fut.await?;
+                        Ok(res)
+                    })
+                }
+            }
         }
     }
 }

src/handlers/http/mod.rs

@@ -39,18 +39,25 @@ pub mod logstream;
 pub mod middleware;
 pub mod modal;
 pub mod oidc;
+pub mod prism_home;
+pub mod prism_logstream;
 pub mod query;
 pub mod rbac;
 pub mod role;
 pub mod users;
 pub const MAX_EVENT_PAYLOAD_SIZE: usize = 10485760;
 pub const API_BASE_PATH: &str = "api";
 pub const API_VERSION: &str = "v1";
+pub const PRISM_BASE_PATH: &str = "prism";
 
 pub fn base_path() -> String {
     format!("/{API_BASE_PATH}/{API_VERSION}")
 }
 
+pub fn prism_base_path() -> String {
+    format!("/{API_BASE_PATH}/{PRISM_BASE_PATH}/{API_VERSION}")
+}
+
 pub fn metrics_path() -> String {
     format!("{}/metrics", base_path())
 }