Updates for Keystone

src/metastore/metastore_traits.rs

@@ -16,7 +16,7 @@
  *
  */
 
-use std::collections::{BTreeMap, HashSet};
+use std::collections::{BTreeMap, HashMap, HashSet};
 
 use arrow_schema::Schema;
 use bytes::Bytes;
@@ -44,6 +44,25 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
     async fn initiate_connection(&self) -> Result<(), MetastoreError>;
     async fn get_objects(&self, parent_path: &str) -> Result<Vec<Bytes>, MetastoreError>;
 
+    /// overview
+    async fn get_overviews(&self) -> Result<HashMap<String, Option<Bytes>>, MetastoreError>;
+    async fn put_overview(
+        &self,
+        obj: &dyn MetastoreObject,
+        stream: &str,
+    ) -> Result<(), MetastoreError>;
+    async fn delete_overview(&self, stream: &str) -> Result<(), MetastoreError>;
+
+    /// keystone
+    async fn get_keystones(&self) -> Result<Vec<Bytes>, MetastoreError>;
+    async fn put_keystone(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError>;
+    async fn delete_keystone(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError>;
+
+    /// conversations
+    async fn get_conversations(&self) -> Result<Vec<Bytes>, MetastoreError>;
+    async fn put_conversation(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError>;
+    async fn delete_conversation(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError>;
+
     /// alerts
     async fn get_alerts(&self) -> Result<Vec<Bytes>, MetastoreError>;
     async fn put_alert(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError>;

src/metastore/metastores/object_store_metastore.rs

@@ -17,7 +17,7 @@
  */
 
 use std::{
-    collections::{BTreeMap, HashSet},
+    collections::{BTreeMap, HashMap, HashSet},
     sync::Arc,
 };
 
@@ -80,6 +80,102 @@ impl Metastore for ObjectStoreMetastore {
             .await?)
     }
 
+    /// This function fetches all the overviews from the underlying object store
+    async fn get_overviews(&self) -> Result<HashMap<String, Option<Bytes>>, MetastoreError> {
+        let streams = self.list_streams().await?;
+
+        let mut all_overviews = HashMap::new();
+        for stream in streams {
+            let overview_path = RelativePathBuf::from_iter([&stream, "overview"]);
+
+            // if the file doesn't exist, load an empty overview
+            let overview = (self.storage.get_object(&overview_path).await).ok();
+
+            all_overviews.insert(stream, overview);
+        }
+
+        Ok(all_overviews)
+    }
+
+    /// This function puts an overview in the object store at the given path
+    async fn put_overview(
+        &self,
+        obj: &dyn MetastoreObject,
+        stream: &str,
+    ) -> Result<(), MetastoreError> {
+        let path = RelativePathBuf::from_iter([stream, "overview"]);
+        Ok(self.storage.put_object(&path, to_bytes(obj)).await?)
+    }
+
+    /// Delete an overview
+    async fn delete_overview(&self, stream: &str) -> Result<(), MetastoreError> {
+        let path = RelativePathBuf::from_iter([stream, "overview"]);
+        Ok(self
+            .storage
+            .delete_object(&path)
+            .await?)
+    }
+
+    /// This function fetches all the keystones from the underlying object store
+    async fn get_keystones(&self) -> Result<Vec<Bytes>, MetastoreError> {
+        let keystone_path = RelativePathBuf::from_iter([".keystone"]);
+        let keystones = self
+            .storage
+            .get_objects(
+                Some(&keystone_path),
+                Box::new(|file_name| {
+                    file_name.ends_with(".json") && !file_name.starts_with("conv_")
+                }),
+            )
+            .await?;
+
+        Ok(keystones)
+    }
+
+    /// This function puts a keystone in the object store at the given path
+    async fn put_keystone(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError> {
+        let id = obj.get_object_id();
+        let path = RelativePathBuf::from_iter([".keystone", &format!("{id}.json")]);
+        Ok(self.storage.put_object(&path, to_bytes(obj)).await?)
+    }
+
+    /// Delete a keystone
+    async fn delete_keystone(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError> {
+        let id = obj.get_object_id();
+        let path = RelativePathBuf::from_iter([".keystone", &format!("{id}.json")]);
+        Ok(self.storage.delete_object(&path).await?)
+    }
+
+    /// This function fetches all the conversations from the underlying object store
+    async fn get_conversations(&self) -> Result<Vec<Bytes>, MetastoreError> {
+        let keystone_path = RelativePathBuf::from_iter([".keystone"]);
+        let conversations = self
+            .storage
+            .get_objects(
+                Some(&keystone_path),
+                Box::new(|file_name| {
+                    file_name.ends_with(".json") && file_name.starts_with("conv_")
+                }),
+            )
+            .await?;
+
+        Ok(conversations)
+    }
+
+    /// This function puts a conversation in the object store at the given path
+    async fn put_conversation(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError> {
+        let id = obj.get_object_id();
+        let path = RelativePathBuf::from_iter([".keystone", &format!("conv_{id}.json")]);
+        Ok(self.storage.put_object(&path, to_bytes(obj)).await?)
+    }
+
+    /// Delete a conversation
+    async fn delete_conversation(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError> {
+        let id = obj.get_object_id();
+        let path = RelativePathBuf::from_iter([".keystone", &format!("conv_{id}.json")]);
+        Ok(self.storage.delete_object(&path).await?)
+    }
+
     /// This function fetches all the alerts from the underlying object store
     async fn get_alerts(&self) -> Result<Vec<Bytes>, MetastoreError> {
         let alerts_path = RelativePathBuf::from(ALERTS_ROOT_DIRECTORY);

src/prism/logstream/mod.rs

@@ -144,7 +144,7 @@ async fn get_stats(stream_name: &str) -> Result<QueriedStats, PrismLogstreamErro
     Ok(stats)
 }
 
-async fn get_stream_info_helper(stream_name: &str) -> Result<StreamInfo, StreamError> {
+pub async fn get_stream_info_helper(stream_name: &str) -> Result<StreamInfo, StreamError> {
     // For query mode, if the stream not found in memory map,
     //check if it exists in the storage
     //create stream and schema from storage

src/utils/error.rs

@@ -0,0 +1,36 @@
+/*
+ * Parseable Server (C) 2022 - 2024 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+use http::StatusCode;
+use serde::Serialize;
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct DetailedError {
+    pub operation: String,
+    pub message: String,
+    pub stream_name: Option<String>,
+    pub timestamp: chrono::DateTime<chrono::Utc>,
+    pub metadata: Option<std::collections::HashMap<String, String>>,
+    pub status_code: u16,
+}
+
+impl DetailedError {
+    pub fn status_code(&self) -> StatusCode {
+        StatusCode::from_u16(self.status_code).unwrap()
+    }
+}

src/utils/mod.rs

@@ -18,6 +18,7 @@
 
 pub mod actix;
 pub mod arrow;
+pub mod error;
 pub mod header_parsing;
 pub mod human_size;
 pub mod json;