fix: s3 get objects

Refactor object storage to filter objects by starts_with_pattern

Author: Eshanatnight

server/src/handlers/http/cluster/mod.rs

@@ -117,7 +117,7 @@ pub async fn fetch_stats_from_ingesters(
     let obs = CONFIG
         .storage()
         .get_object_store()
-        .get_objects(Some(&path))
+        .get_objects(Some(&path), ".ingester")
         .await?;
     let mut ingestion_size = 0u64;
     let mut storage_size = 0u64;
@@ -346,7 +346,7 @@ pub async fn get_ingester_info() -> anyhow::Result<IngesterMetadataArr> {
 
     let root_path = RelativePathBuf::from(PARSEABLE_ROOT_DIRECTORY);
     let arr = store
-        .get_objects(Some(&root_path))
+        .get_objects(Some(&root_path), "ingester")
         .await?
         .iter()
         // this unwrap will most definateley shoot me in the foot later

server/src/handlers/http/modal/ingest_server.rs

@@ -251,7 +251,7 @@ impl IngestServer {
         let store = CONFIG.storage().get_object_store();
         let base_path = RelativePathBuf::from("");
         let ingester_metadata = store
-            .get_objects(Some(&base_path))
+            .get_objects(Some(&base_path), "ingester")
             .await?
             .iter()
             // this unwrap will most definateley shoot me in the foot later

server/src/storage/localfs.rs

@@ -189,9 +189,11 @@ impl ObjectStorage for LocalFS {
         Ok(path_arr)
     }
 
+    /// currently it is not using the starts_with_pattern
     async fn get_objects(
         &self,
         base_path: Option<&RelativePath>,
+        _starts_with_pattern: &str
     ) -> Result<Vec<Bytes>, ObjectStorageError> {
         let time = Instant::now();
 

server/src/storage/object_storage.rs

@@ -69,6 +69,7 @@ pub trait ObjectStorage: Sync + 'static {
     async fn get_objects(
         &self,
         base_path: Option<&RelativePath>,
+        starts_with_pattern: &str
     ) -> Result<Vec<Bytes>, ObjectStorageError>;
     async fn put_object(
         &self,

server/src/storage/s3.rs

@@ -416,6 +416,7 @@ impl ObjectStorage for S3 {
     async fn get_objects(
         &self,
         base_path: Option<&RelativePath>,
+        starts_with_pattern: &str,
     ) -> Result<Vec<Bytes>, ObjectStorageError> {
         let instant = Instant::now();
 
@@ -430,7 +431,7 @@ impl ObjectStorage for S3 {
         let mut res = vec![];
 
         while let Some(meta) = list_stream.next().await.transpose()? {
-            let ingester_file = meta.location.filename().unwrap().starts_with("ingester");
+            let ingester_file = meta.location.filename().unwrap().starts_with(starts_with_pattern);
 
             if !ingester_file {
                 continue;