refactor: DRY get_all_saved_filters

Devdutt Shenoi · Devdutt Shenoi · commit d08cd828074b · 2025-02-01T19:09:00.000+05:30
diff --git a/src/storage/azure_blob.rs b/src/storage/azure_blob.rs
@@ -653,7 +653,10 @@ impl ObjectStorage for BlobStore {
             .collect::<Vec<_>>())
     }
 
-    async fn list_dirs_relative(&self, relative_path: &RelativePath) -> Result<Vec<String>, ObjectStorageError> {
+    async fn list_dirs_relative(
+        &self,
+        relative_path: &RelativePath,
+    ) -> Result<Vec<String>, ObjectStorageError> {
         let prefix = object_store::path::Path::from(relative_path.as_str());
         let resp = self.client.list_with_delimiter(Some(&prefix)).await?;
 
@@ -701,53 +704,6 @@ impl ObjectStorage for BlobStore {
         Ok(dashboards)
     }
 
-    async fn get_all_saved_filters(
-        &self,
-    ) -> Result<HashMap<RelativePathBuf, Vec<Bytes>>, ObjectStorageError> {
-        let mut filters: HashMap<RelativePathBuf, Vec<Bytes>> = HashMap::new();
-        let users_root_path = object_store::path::Path::from(USERS_ROOT_DIR);
-        let resp = self
-            .client
-            .list_with_delimiter(Some(&users_root_path))
-            .await?;
-
-        let users = resp
-            .common_prefixes
-            .iter()
-            .flat_map(|path| path.parts())
-            .filter(|name| name.as_ref() != USERS_ROOT_DIR)
-            .map(|name| name.as_ref().to_string())
-            .collect::<Vec<_>>();
-        for user in users {
-            let user_filters_path =
-                object_store::path::Path::from(format!("{USERS_ROOT_DIR}/{user}/filters",));
-            let resp = self
-                .client
-                .list_with_delimiter(Some(&user_filters_path))
-                .await?;
-            let streams = resp
-                .common_prefixes
-                .iter()
-                .filter(|name| name.as_ref() != USERS_ROOT_DIR)
-                .map(|name| name.as_ref().to_string())
-                .collect::<Vec<_>>();
-            for stream in streams {
-                let filters_path = RelativePathBuf::from(&stream);
-                let filter_bytes = self
-                    .get_objects(
-                        Some(&filters_path),
-                        Box::new(|file_name| file_name.ends_with(".json")),
-                    )
-                    .await?;
-                filters
-                    .entry(filters_path)
-                    .or_default()
-                    .extend(filter_bytes);
-            }
-        }
-        Ok(filters)
-    }
-
     ///fetch all correlations uploaded in object store
     /// return the correlation file path and all correlation json bytes for each file path
     async fn get_all_correlations(
diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs
@@ -353,8 +353,10 @@ impl ObjectStorage for LocalFS {
         Ok(dirs)
     }
 
-    
-    async fn list_dirs_relative(&self, relative_path: &RelativePath) -> Result<Vec<String>, ObjectStorageError> {
+    async fn list_dirs_relative(
+        &self,
+        relative_path: &RelativePath,
+    ) -> Result<Vec<String>, ObjectStorageError> {
         let root = self.root.join(relative_path.as_str());
         let dirs = ReadDirStream::new(fs::read_dir(root).await?)
             .try_collect::<Vec<DirEntry>>()
@@ -368,7 +370,7 @@ impl ObjectStorage for LocalFS {
             .into_iter()
             .flatten()
             .collect::<Vec<_>>();
-        
+
         Ok(dirs)
     }
 
@@ -402,47 +404,6 @@ impl ObjectStorage for LocalFS {
         Ok(dashboards)
     }
 
-    async fn get_all_saved_filters(
-        &self,
-    ) -> Result<HashMap<RelativePathBuf, Vec<Bytes>>, ObjectStorageError> {
-        let mut filters: HashMap<RelativePathBuf, Vec<Bytes>> = HashMap::new();
-        let users_root_path = self.root.join(USERS_ROOT_DIR);
-        let directories = ReadDirStream::new(fs::read_dir(&users_root_path).await?);
-        let users: Vec<DirEntry> = directories.try_collect().await?;
-        for user in users {
-            if !user.path().is_dir() {
-                continue;
-            }
-            let stream_root_path = users_root_path.join(user.path()).join("filters");
-            let directories = ReadDirStream::new(fs::read_dir(&stream_root_path).await?);
-            let streams: Vec<DirEntry> = directories.try_collect().await?;
-            for stream in streams {
-                if !stream.path().is_dir() {
-                    continue;
-                }
-                let filters_path = users_root_path
-                    .join(user.path())
-                    .join("filters")
-                    .join(stream.path());
-                let directories = ReadDirStream::new(fs::read_dir(&filters_path).await?);
-                let filters_files: Vec<DirEntry> = directories.try_collect().await?;
-                for filter in filters_files {
-                    let filter_absolute_path = filter.path();
-                    let file = fs::read(filter_absolute_path.clone()).await?;
-                    let filter_relative_path = filter_absolute_path
-                        .strip_prefix(self.root.as_path())
-                        .unwrap();
-
-                    filters
-                        .entry(RelativePathBuf::from_path(filter_relative_path).unwrap())
-                        .or_default()
-                        .push(file.into());
-                }
-            }
-        }
-        Ok(filters)
-    }
-
     ///fetch all correlations stored in disk
     /// return the correlation file path and all correlation json bytes for each file path
     async fn get_all_correlations(
diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs
@@ -94,10 +94,36 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
     async fn list_streams(&self) -> Result<HashSet<LogStream>, ObjectStorageError>;
     async fn list_old_streams(&self) -> Result<HashSet<LogStream>, ObjectStorageError>;
     async fn list_dirs(&self) -> Result<Vec<String>, ObjectStorageError>;
-    async fn list_dirs_relative(&self, relative_path: &RelativePath) -> Result<Vec<String>, ObjectStorageError>;
+    async fn list_dirs_relative(
+        &self,
+        relative_path: &RelativePath,
+    ) -> Result<Vec<String>, ObjectStorageError>;
+
     async fn get_all_saved_filters(
         &self,
-    ) -> Result<HashMap<RelativePathBuf, Vec<Bytes>>, ObjectStorageError>;
+    ) -> Result<HashMap<RelativePathBuf, Vec<Bytes>>, ObjectStorageError> {
+        let mut filters: HashMap<RelativePathBuf, Vec<Bytes>> = HashMap::new();
+
+        let users_dir = RelativePathBuf::from_iter([USERS_ROOT_DIR]);
+        for user in self.list_dirs_relative(&users_dir).await? {
+            let stream_dir = RelativePathBuf::from_iter([USERS_ROOT_DIR, &user]);
+            for stream in self.list_dirs_relative(&stream_dir).await? {
+                let filters_path = RelativePathBuf::from(&stream);
+                let filter_bytes = self
+                    .get_objects(
+                        Some(&filters_path),
+                        Box::new(|file_name| file_name.ends_with(".json")),
+                    )
+                    .await?;
+                filters
+                    .entry(filters_path)
+                    .or_default()
+                    .extend(filter_bytes);
+            }
+        }
+        Ok(filters)
+    }
+
     async fn get_all_dashboards(
         &self,
     ) -> Result<HashMap<RelativePathBuf, Vec<Bytes>>, ObjectStorageError>;
diff --git a/src/storage/s3.rs b/src/storage/s3.rs
@@ -786,7 +786,10 @@ impl ObjectStorage for S3 {
             .collect::<Vec<_>>())
     }
 
-    async fn list_dirs_relative(&self, relative_path: &RelativePath) -> Result<Vec<String>, ObjectStorageError> {
+    async fn list_dirs_relative(
+        &self,
+        relative_path: &RelativePath,
+    ) -> Result<Vec<String>, ObjectStorageError> {
         let prefix = object_store::path::Path::from(relative_path.as_str());
         let resp = self.client.list_with_delimiter(Some(&prefix)).await?;
 
@@ -834,53 +837,6 @@ impl ObjectStorage for S3 {
         Ok(dashboards)
     }
 
-    async fn get_all_saved_filters(
-        &self,
-    ) -> Result<HashMap<RelativePathBuf, Vec<Bytes>>, ObjectStorageError> {
-        let mut filters: HashMap<RelativePathBuf, Vec<Bytes>> = HashMap::new();
-        let users_root_path = object_store::path::Path::from(USERS_ROOT_DIR);
-        let resp = self
-            .client
-            .list_with_delimiter(Some(&users_root_path))
-            .await?;
-
-        let users = resp
-            .common_prefixes
-            .iter()
-            .flat_map(|path| path.parts())
-            .filter(|name| name.as_ref() != USERS_ROOT_DIR)
-            .map(|name| name.as_ref().to_string())
-            .collect::<Vec<_>>();
-        for user in users {
-            let user_filters_path =
-                object_store::path::Path::from(format!("{USERS_ROOT_DIR}/{user}/filters",));
-            let resp = self
-                .client
-                .list_with_delimiter(Some(&user_filters_path))
-                .await?;
-            let streams = resp
-                .common_prefixes
-                .iter()
-                .filter(|name| name.as_ref() != USERS_ROOT_DIR)
-                .map(|name| name.as_ref().to_string())
-                .collect::<Vec<_>>();
-            for stream in streams {
-                let filters_path = RelativePathBuf::from(&stream);
-                let filter_bytes = self
-                    .get_objects(
-                        Some(&filters_path),
-                        Box::new(|file_name| file_name.ends_with(".json")),
-                    )
-                    .await?;
-                filters
-                    .entry(filters_path)
-                    .or_default()
-                    .extend(filter_bytes);
-            }
-        }
-        Ok(filters)
-    }
-
     ///fetch all correlations stored in object store
     /// return the correlation file path and all correlation json bytes for each file path
     async fn get_all_correlations(
