Changes for stream.json

Author: parmesant

src/alerts/alert_structs.rs

@@ -531,11 +531,11 @@ pub struct NotificationStateRequest {
 }
 
 impl MetastoreObject for AlertConfig {
-    fn get_id(&self) -> String {
+    fn get_object_id(&self) -> String {
         self.id.to_string()
     }
 
-    fn get_path(&self) -> String {
+    fn get_object_path(&self) -> String {
         alert_json_path(self.id).to_string()
     }
 }

src/alerts/alert_traits.rs

@@ -22,6 +22,7 @@ use crate::{
         alert_enums::NotificationState,
         alert_structs::{Context, ThresholdConfig},
     },
+    metastore::metastore_traits::MetastoreObject,
     rbac::map::SessionKey,
 };
 use chrono::{DateTime, Utc};
@@ -47,7 +48,7 @@ pub trait MessageCreation {
 }
 
 #[async_trait]
-pub trait AlertTrait: Debug + Send + Sync {
+pub trait AlertTrait: Debug + Send + Sync + MetastoreObject {
     async fn eval_alert(&self) -> Result<Option<String>, AlertError>;
     async fn validate(&self, session_key: &SessionKey) -> Result<(), AlertError>;
     async fn update_notification_state(

src/alerts/alert_types.rs

@@ -35,9 +35,11 @@ use crate::{
         target::{self, NotificationConfig},
     },
     handlers::http::query::create_streams_for_distributed,
+    metastore::metastore_traits::MetastoreObject,
     parseable::PARSEABLE,
     query::resolve_stream_names,
     rbac::map::SessionKey,
+    storage::object_storage::alert_json_path,
     utils::user_auth_for_query,
 };
 
@@ -65,6 +67,16 @@ pub struct ThresholdAlert {
     pub last_triggered_at: Option<DateTime<Utc>>,
 }
 
+impl MetastoreObject for ThresholdAlert {
+    fn get_object_path(&self) -> String {
+        alert_json_path(self.id).to_string()
+    }
+
+    fn get_object_id(&self) -> String {
+        self.id.to_string()
+    }
+}
+
 #[async_trait]
 impl AlertTrait for ThresholdAlert {
     async fn eval_alert(&self) -> Result<Option<String>, AlertError> {

src/catalog/mod.rs

@@ -114,8 +114,13 @@ pub async fn update_snapshot(
         return Ok(());
     }
 
-    let mut meta = storage.get_object_store_format(stream_name).await?;
-
+    let mut meta: ObjectStoreFormat = serde_json::from_slice(
+        &PARSEABLE
+            .metastore
+            .get_stream_json(stream_name, false)
+            .await
+            .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
+    )?;
     let partition_groups = group_changes_by_partition(changes, &meta.time_partition);
 
     let new_manifest_entries =
@@ -458,7 +463,14 @@ pub async fn remove_manifest_from_snapshot(
 ) -> Result<(), ObjectStorageError> {
     if !dates.is_empty() {
         // get current snapshot
-        let mut meta = storage.get_object_store_format(stream_name).await?;
+        let mut meta: ObjectStoreFormat = serde_json::from_slice(
+            &PARSEABLE
+                .metastore
+                .get_stream_json(stream_name, false)
+                .await
+                .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
+        )?;
+
         let meta_for_stats = meta.clone();
         update_deleted_stats(storage.clone(), stream_name, meta_for_stats, dates.clone()).await?;
         let manifests = &mut meta.snapshot.manifest_list;

src/enterprise/utils.rs

@@ -15,7 +15,7 @@ use crate::{
     event,
     parseable::PARSEABLE,
     query::{PartialTimeFilter, stream_schema_provider::ManifestExt},
-    storage::{ObjectStorage, ObjectStorageError, ObjectStoreFormat, STREAM_ROOT_DIRECTORY},
+    storage::{ObjectStorage, ObjectStorageError, ObjectStoreFormat},
     utils::time::TimeRange,
 };
 
@@ -68,7 +68,13 @@ pub async fn fetch_parquet_file_paths(
 ) -> 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 object_store_format: ObjectStoreFormat = serde_json::from_slice(
+        &PARSEABLE
+            .metastore
+            .get_stream_json(stream, false)
+            .await
+            .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?,
+    )?;
 
     let time_partition = object_store_format.time_partition;
 
@@ -78,13 +84,7 @@ pub async fn fetch_parquet_file_paths(
 
     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;
+    let obs = PARSEABLE.metastore.get_all_stream_jsons(stream, None).await;
     if let Ok(obs) = obs {
         for ob in obs {
             if let Ok(object_store_format) = serde_json::from_slice::<ObjectStoreFormat>(&ob) {

src/handlers/http/alerts.rs

@@ -262,10 +262,7 @@ pub async fn delete(req: HttpRequest, alert_id: Path<Ulid>) -> Result<impl Respo
     // validate that the user has access to the tables mentioned in the query
     user_auth_for_query(&session_key, alert.get_query()).await?;
 
-    PARSEABLE
-        .metastore
-        .delete_object(alert_json_path(alert_id).as_ref())
-        .await?;
+    PARSEABLE.metastore.delete_alert(&*alert).await?;
 
     // delete from memory
     alerts.delete(alert_id).await?;

src/handlers/http/cluster/mod.rs

@@ -45,13 +45,13 @@ use crate::INTRA_CLUSTER_CLIENT;
 use crate::handlers::http::ingest::ingest_internal_stream;
 use crate::handlers::http::query::{Query, QueryError, TIME_ELAPSED_HEADER};
 use crate::metrics::prom_utils::Metrics;
+use crate::option::Mode;
 use crate::parseable::PARSEABLE;
 use crate::rbac::role::model::DefaultPrivilege;
 use crate::rbac::user::User;
 use crate::stats::Stats;
 use crate::storage::{
     ObjectStorage, ObjectStorageError, ObjectStoreFormat, PARSEABLE_ROOT_DIRECTORY,
-    STREAM_ROOT_DIRECTORY,
 };
 
 use super::base_path_without_preceding_slash;
@@ -497,16 +497,9 @@ pub fn fetch_daily_stats(
 pub async fn fetch_stats_from_ingestors(
     stream_name: &str,
 ) -> Result<Vec<utils::QueriedStats>, StreamError> {
-    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")
-            }),
-        )
+        .metastore
+        .get_all_stream_jsons(stream_name, Some(Mode::Ingest))
         .await?;
 
     let mut ingestion_size = 0u64;

src/handlers/http/logstream.rs

@@ -28,7 +28,7 @@ use crate::rbac::Users;
 use crate::rbac::role::Action;
 use crate::stats::{Stats, event_labels_date, storage_size_labels_date};
 use crate::storage::retention::Retention;
-use crate::storage::{StreamInfo, StreamType};
+use crate::storage::{ObjectStoreFormat, StreamInfo, StreamType};
 use crate::utils::actix::extract_session_key_from_req;
 use crate::utils::json::flatten::{
     self, convert_to_array, generic_flattening, has_more_than_max_allowed_levels,
@@ -413,7 +413,12 @@ pub async fn put_stream_hot_tier(
         .put_hot_tier(&stream_name, &mut hottier)
         .await?;
     let storage = PARSEABLE.storage().get_object_store();
-    let mut stream_metadata = storage.get_object_store_format(&stream_name).await?;
+    let mut stream_metadata: ObjectStoreFormat = serde_json::from_slice(
+        &PARSEABLE
+            .metastore
+            .get_stream_json(&stream_name, false)
+            .await?,
+    )?;
     stream_metadata.hot_tier_enabled = true;
     storage
         .put_stream_manifest(&stream_name, &stream_metadata)
@@ -491,6 +496,7 @@ pub mod error {
 
     use crate::{
         hottier::HotTierError,
+        metastore::MetastoreError,
         parseable::StreamNotFound,
         storage::ObjectStorageError,
         validator::error::{
@@ -563,6 +569,8 @@ pub mod error {
         HotTierError(#[from] HotTierError),
         #[error("Invalid query parameter: {0}")]
         InvalidQueryParameter(String),
+        #[error("{0:?}")]
+        MetastoreError(#[from] MetastoreError),
     }
 
     impl actix_web::ResponseError for StreamError {
@@ -599,13 +607,21 @@ pub mod error {
                 StreamError::HotTierValidation(_) => StatusCode::BAD_REQUEST,
                 StreamError::HotTierError(_) => StatusCode::INTERNAL_SERVER_ERROR,
                 StreamError::InvalidQueryParameter(_) => StatusCode::BAD_REQUEST,
+                StreamError::MetastoreError(e) => e.status_code(),
             }
         }
 
         fn error_response(&self) -> actix_web::HttpResponse<actix_web::body::BoxBody> {
-            actix_web::HttpResponse::build(self.status_code())
-                .insert_header(ContentType::plaintext())
-                .body(self.to_string())
+            match self {
+                StreamError::MetastoreError(metastore_error) => {
+                    actix_web::HttpResponse::build(metastore_error.status_code())
+                        .insert_header(ContentType::json())
+                        .json(metastore_error.to_detail())
+                }
+                _ => actix_web::HttpResponse::build(self.status_code())
+                    .insert_header(ContentType::plaintext())
+                    .body(self.to_string()),
+            }
         }
     }
 }

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

@@ -26,7 +26,6 @@ use actix_web::{
 use bytes::Bytes;
 use chrono::Utc;
 use http::StatusCode;
-use relative_path::RelativePathBuf;
 use tokio::sync::Mutex;
 use tracing::{error, warn};
 
@@ -48,7 +47,7 @@ use crate::{
     hottier::HotTierManager,
     parseable::{PARSEABLE, StreamNotFound},
     stats,
-    storage::{ObjectStoreFormat, STREAM_ROOT_DIRECTORY, StreamType},
+    storage::{ObjectStoreFormat, StreamType},
 };
 const STATS_DATE_QUERY_PARAM: &str = "date";
 
@@ -165,14 +164,9 @@ pub async fn get_stats(
 
         if !date_value.is_empty() {
             // 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.ends_with("stream.json")),
-                )
+                .metastore
+                .get_all_stream_jsons(&stream_name, None)
                 .await?;
 
             let mut stream_jsons = Vec::new();

src/handlers/http/query.rs

@@ -18,6 +18,7 @@
 
 use crate::event::error::EventError;
 use crate::handlers::http::fetch_schema;
+use crate::metastore::MetastoreError;
 use crate::option::Mode;
 use crate::rbac::map::SessionKey;
 use crate::utils::arrow::record_batches_to_json;
@@ -578,12 +579,15 @@ Description: {0}"#
     NoAvailableQuerier,
     #[error("{0}")]
     ParserError(#[from] ParserError),
+    #[error("{0:?}")]
+    MetastoreError(#[from] MetastoreError),
 }
 
 impl actix_web::ResponseError for QueryError {
     fn status_code(&self) -> http::StatusCode {
         match self {
             QueryError::Execute(_) | QueryError::JsonParse(_) => StatusCode::INTERNAL_SERVER_ERROR,
+            QueryError::MetastoreError(e) => e.status_code(),
             _ => StatusCode::BAD_REQUEST,
         }
     }