Ported the remaining structs over to metastore

Author: parmesant

src/alerts/mod.rs

@@ -940,7 +940,7 @@ pub enum AlertError {
     Unimplemented(String),
     #[error("{0}")]
     ValidationFailure(String),
-    #[error("{0}")]
+    #[error(transparent)]
     MetastoreError(#[from] MetastoreError),
 }
 
@@ -1237,8 +1237,6 @@ impl AlertManagerTrait for Alerts {
         alert_id: Ulid,
         new_notification_state: NotificationState,
     ) -> Result<(), AlertError> {
-        // let store = PARSEABLE.storage.get_object_store();
-
         // read and modify alert
         let mut write_access = self.alerts.write().await;
         let mut alert: Box<dyn AlertTrait> = if let Some(alert) = write_access.get(&alert_id) {

src/alerts/target.rs

@@ -24,7 +24,6 @@ use std::{
 
 use async_trait::async_trait;
 use base64::Engine;
-use bytes::Bytes;
 use chrono::Utc;
 use http::{HeaderMap, HeaderValue, header::AUTHORIZATION};
 use itertools::Itertools;
@@ -38,6 +37,7 @@ use url::Url;
 
 use crate::{
     alerts::{AlertError, AlertState, Context, alert_traits::CallableTarget},
+    metastore::metastore_traits::MetastoreObject,
     parseable::PARSEABLE,
     storage::object_storage::target_json_path,
 };
@@ -57,10 +57,9 @@ impl TargetConfigs {
     /// Loads alerts from disk, blocks
     pub async fn load(&self) -> anyhow::Result<()> {
         let mut map = self.target_configs.write().await;
-        let store = PARSEABLE.storage.get_object_store();
 
-        for alert in store.get_targets().await.unwrap_or_default() {
-            map.insert(alert.id, alert);
+        for target in PARSEABLE.metastore.get_targets().await.unwrap_or_default() {
+            map.insert(target.id, target);
         }
 
         Ok(())
@@ -69,12 +68,7 @@ impl TargetConfigs {
     pub async fn update(&self, target: Target) -> Result<(), AlertError> {
         let mut map = self.target_configs.write().await;
         map.insert(target.id, target.clone());
-
-        let path = target_json_path(&target.id);
-
-        let store = PARSEABLE.storage.get_object_store();
-        let target_bytes = serde_json::to_vec(&target)?;
-        store.put_object(&path, Bytes::from(target_bytes)).await?;
+        PARSEABLE.metastore.put_target(&target).await?;
         Ok(())
     }
 
@@ -121,9 +115,7 @@ impl TargetConfigs {
             .await
             .remove(target_id)
             .ok_or(AlertError::InvalidTargetID(target_id.to_string()))?;
-        let path = target_json_path(&target.id);
-        let store = PARSEABLE.storage.get_object_store();
-        store.delete_object(&path).await?;
+        PARSEABLE.metastore.delete_target(&target).await?;
         Ok(target)
     }
 }
@@ -340,6 +332,16 @@ impl Target {
     }
 }
 
+impl MetastoreObject for Target {
+    fn get_object_path(&self) -> String {
+        target_json_path(&self.id).to_string()
+    }
+
+    fn get_object_id(&self) -> String {
+        self.id.to_string()
+    }
+}
+
 fn call_target(target: TargetType, context: Context) {
     trace!("Calling target with context- {context:?}");
     tokio::spawn(async move { target.call(&context).await });

src/catalog/manifest.rs

@@ -21,6 +21,8 @@ use std::collections::HashMap;
 use itertools::Itertools;
 use parquet::{file::reader::FileReader, format::SortingColumn};
 
+use crate::metastore::metastore_traits::MetastoreObject;
+
 use super::column::Column;
 
 #[derive(
@@ -88,6 +90,16 @@ impl Manifest {
     }
 }
 
+impl MetastoreObject for Manifest {
+    fn get_object_path(&self) -> String {
+        unimplemented!()
+    }
+
+    fn get_object_id(&self) -> String {
+        unimplemented!()
+    }
+}
+
 pub fn create_from_parquet_file(
     object_store_path: String,
     fs_file_path: &std::path::Path,

src/catalog/mod.rs

@@ -106,7 +106,6 @@ fn get_file_bounds(
 }
 
 pub async fn update_snapshot(
-    storage: Arc<dyn ObjectStorage>,
     stream_name: &str,
     changes: Vec<manifest::File>,
 ) -> Result<(), ObjectStorageError> {
@@ -124,9 +123,9 @@ pub async fn update_snapshot(
     let partition_groups = group_changes_by_partition(changes, &meta.time_partition);
 
     let new_manifest_entries =
-        process_partition_groups(partition_groups, &mut meta, storage.clone(), stream_name).await?;
+        process_partition_groups(partition_groups, &mut meta, stream_name).await?;
 
-    finalize_snapshot_update(meta, new_manifest_entries, storage, stream_name).await
+    finalize_snapshot_update(meta, new_manifest_entries, stream_name).await
 }
 
 /// Groups manifest file changes by time partitions using Rayon for parallel processing
@@ -214,7 +213,6 @@ fn extract_partition_metrics(stream_name: &str, partition_lower: DateTime<Utc>)
 async fn process_partition_groups(
     partition_groups: HashMap<(DateTime<Utc>, DateTime<Utc>), Vec<manifest::File>>,
     meta: &mut ObjectStoreFormat,
-    storage: Arc<dyn ObjectStorage>,
     stream_name: &str,
 ) -> Result<Vec<snapshot::ManifestItem>, ObjectStorageError> {
     let mut new_manifest_entries = Vec::new();
@@ -227,7 +225,6 @@ async fn process_partition_groups(
             partition_lower,
             partition_changes,
             meta,
-            storage.clone(),
             stream_name,
             events_ingested,
             ingestion_size,
@@ -249,7 +246,6 @@ async fn process_single_partition(
     partition_lower: DateTime<Utc>,
     partition_changes: Vec<manifest::File>,
     meta: &mut ObjectStoreFormat,
-    storage: Arc<dyn ObjectStorage>,
     stream_name: &str,
     events_ingested: u64,
     ingestion_size: u64,
@@ -263,7 +259,6 @@ async fn process_single_partition(
         handle_existing_partition(
             pos,
             partition_changes,
-            storage,
             stream_name,
             meta,
             events_ingested,
@@ -277,7 +272,6 @@ async fn process_single_partition(
         create_manifest(
             partition_lower,
             partition_changes,
-            storage,
             stream_name,
             false,
             meta.clone(),
@@ -294,7 +288,6 @@ async fn process_single_partition(
 async fn handle_existing_partition(
     pos: usize,
     partition_changes: Vec<manifest::File>,
-    storage: Arc<dyn ObjectStorage>,
     stream_name: &str,
     meta: &mut ObjectStoreFormat,
     events_ingested: u64,
@@ -303,22 +296,35 @@ async fn handle_existing_partition(
     partition_lower: DateTime<Utc>,
 ) -> Result<Option<snapshot::ManifestItem>, ObjectStorageError> {
     let manifests = &mut meta.snapshot.manifest_list;
-    let path = partition_path(
-        stream_name,
-        manifests[pos].time_lower_bound,
-        manifests[pos].time_upper_bound,
-    );
 
     let manifest_file_name = manifest_path("").to_string();
     let should_update = manifests[pos].manifest_path.contains(&manifest_file_name);
 
     if should_update {
-        if let Some(mut manifest) = storage.get_manifest(&path).await? {
+        if let Some(mut manifest) = PARSEABLE
+            .metastore
+            .get_manifest(
+                stream_name,
+                manifests[pos].time_lower_bound,
+                manifests[pos].time_upper_bound,
+            )
+            .await
+            .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?
+        {
             // Update existing manifest
             for change in partition_changes {
                 manifest.apply_change(change);
             }
-            storage.put_manifest(&path, manifest).await?;
+            PARSEABLE
+                .metastore
+                .put_manifest(
+                    &manifest,
+                    stream_name,
+                    manifests[pos].time_lower_bound,
+                    manifests[pos].time_upper_bound,
+                )
+                .await
+                .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?;
 
             manifests[pos].events_ingested = events_ingested;
             manifests[pos].ingestion_size = ingestion_size;
@@ -329,7 +335,6 @@ async fn handle_existing_partition(
             create_manifest(
                 partition_lower,
                 partition_changes,
-                storage,
                 stream_name,
                 false,
                 meta.clone(),
@@ -344,7 +349,6 @@ async fn handle_existing_partition(
         create_manifest(
             partition_lower,
             partition_changes,
-            storage,
             stream_name,
             false,
             ObjectStoreFormat::default(),
@@ -360,7 +364,6 @@ async fn handle_existing_partition(
 async fn finalize_snapshot_update(
     mut meta: ObjectStoreFormat,
     new_manifest_entries: Vec<snapshot::ManifestItem>,
-    storage: Arc<dyn ObjectStorage>,
     stream_name: &str,
 ) -> Result<(), ObjectStorageError> {
     // Add all new manifest entries to the snapshot
@@ -370,15 +373,18 @@ async fn finalize_snapshot_update(
     if let Some(stats) = stats {
         meta.stats = stats;
     }
-    storage.put_stream_manifest(stream_name, &meta).await?;
+    PARSEABLE
+        .metastore
+        .put_stream_json(&meta, stream_name)
+        .await
+        .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?;
     Ok(())
 }
 
 #[allow(clippy::too_many_arguments)]
 async fn create_manifest(
     lower_bound: DateTime<Utc>,
     changes: Vec<manifest::File>,
-    storage: Arc<dyn ObjectStorage>,
     stream_name: &str,
     update_snapshot: bool,
     mut meta: ObjectStoreFormat,
@@ -424,15 +430,19 @@ async fn create_manifest(
         }
     }
 
-    let manifest_file_name = manifest_path("").to_string();
-    let path = partition_path(stream_name, lower_bound, upper_bound).join(&manifest_file_name);
-    storage
-        .put_object(&path, serde_json::to_vec(&manifest)?.into())
-        .await?;
+    PARSEABLE
+        .metastore
+        .put_manifest(&manifest, stream_name, lower_bound, upper_bound)
+        .await
+        .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?;
 
-    let path_url = storage.absolute_url(&path);
+    let path_url = &PARSEABLE
+        .metastore
+        .get_manifest_path(stream_name, lower_bound, upper_bound)
+        .await
+        .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?;
     let new_snapshot_entry = snapshot::ManifestItem {
-        manifest_path: path_url.to_string(),
+        manifest_path: path_url.to_owned(),
         time_lower_bound: lower_bound,
         time_upper_bound: upper_bound,
         events_ingested,
@@ -449,7 +459,13 @@ async fn create_manifest(
             meta.stats = stats;
         }
         meta.first_event_at = first_event_at;
-        storage.put_stream_manifest(stream_name, &meta).await?;
+
+        PARSEABLE
+            .metastore
+            .put_stream_json(&meta, stream_name)
+            .await
+            .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?;
+
         Ok(None)
     } else {
         Ok(Some(new_snapshot_entry))