chore: refactor logic for deleting zombie filters

Author: spuckhafte

src/handlers/http/logstream.rs

@@ -29,7 +29,6 @@ use crate::rbac::role::Action;
 use crate::stats::{Stats, event_labels_date, storage_size_labels_date};
 use crate::storage::retention::Retention;
 use crate::storage::{ObjectStoreFormat, StreamInfo, StreamType};
-use crate::users::filters::{FILTERS, Filter};
 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,
@@ -57,21 +56,6 @@ pub async fn delete(stream_name: Path<String>) -> Result<impl Responder, StreamE
 
     let objectstore = PARSEABLE.storage.get_object_store();
 
-    let all_filters = PARSEABLE.metastore.get_filters().await?;
-    // collect filters associated with the logstream being deleted
-    let filters_for_stream: Vec<Filter> = all_filters
-        .into_iter()
-        .filter(|filter| filter.stream_name == stream_name)
-        .collect();
-
-    for filter in filters_for_stream.iter() {
-        PARSEABLE.metastore.delete_filter(filter).await?;
-        
-        if let Some(filter_id) = filter.filter_id.as_ref() {
-            FILTERS.delete_filter(filter_id).await;
-        }
-    }
-
     // Delete from storage
     objectstore.delete_stream(&stream_name).await?;
     // Delete from staging
@@ -95,6 +79,12 @@ pub async fn delete(stream_name: Path<String>) -> Result<impl Responder, StreamE
     stats::delete_stats(&stream_name, "json")
         .unwrap_or_else(|e| warn!("failed to delete stats for stream {}: {:?}", stream_name, e));
 
+
+    // clear filters associated to the deleted logstream
+    if let Err(e) = PARSEABLE.metastore.delete_zombie_filters(&stream_name).await {
+        warn!("failed to delete zombie filters associated to stream {}: {:?}", stream_name, e);
+    }
+
     Ok((format!("log stream {stream_name} deleted"), StatusCode::OK))
 }
 

src/metastore/metastore_traits.rs

@@ -108,6 +108,7 @@ pub trait Metastore: std::fmt::Debug + Send + Sync {
     async fn get_filters(&self) -> Result<Vec<Filter>, MetastoreError>;
     async fn put_filter(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError>;
     async fn delete_filter(&self, obj: &dyn MetastoreObject) -> Result<(), MetastoreError>;
+    async fn delete_zombie_filters(&self, stream_name: &str) -> Result<bool, MetastoreError>;
 
     /// correlations
     async fn get_correlations(&self) -> Result<Vec<Bytes>, MetastoreError>;

src/metastore/metastores/object_store_metastore.rs

@@ -56,7 +56,7 @@ use crate::{
             parseable_json_path, schema_path, stream_json_path, to_bytes,
         },
     },
-    users::filters::{Filter, migrate_v1_v2},
+    users::filters::{FILTERS, Filter, migrate_v1_v2},
 };
 
 /// Using PARSEABLE's storage as a metastore (default)
@@ -546,6 +546,38 @@ impl Metastore for ObjectStoreMetastore {
             .await?)
     }
 
+    // clear filters associated to a deleted stream
+    async fn delete_zombie_filters(&self, stream_name: &str) -> Result<bool, MetastoreError> {
+        // stream should not exist in order to have zombie filters
+        if PARSEABLE.check_stream_exists(stream_name) {
+            warn!("no zombie filters cleared for [undeleted] stream {}", stream_name);
+            return Ok(false);
+        }
+
+        let all_filters = match PARSEABLE.metastore.get_filters().await {
+            Ok(all_f) => all_f,
+            Err(e) => {
+                return Err(e);
+            }
+        };
+
+        // collect filters associated with the logstream being deleted
+        let filters_for_stream: Vec<Filter> = all_filters
+            .into_iter()
+            .filter(|filter| filter.stream_name == stream_name)
+            .collect();
+
+        for filter in filters_for_stream.iter() {
+            PARSEABLE.metastore.delete_filter(filter).await?;
+            
+            if let Some(filter_id) = filter.filter_id.as_ref() {
+                FILTERS.delete_filter(filter_id).await;
+            }
+        }
+
+        return Ok(true);
+    }
+
     /// Get all correlations
     async fn get_correlations(&self) -> Result<Vec<Bytes>, MetastoreError> {
         let mut correlations = Vec::new();

src/parseable/mod.rs

@@ -235,6 +235,15 @@ impl Parseable {
                 .unwrap_or_default()
     }
 
+    // check if a stream exists
+    pub fn check_stream_exists(&self, stream_name: &str) -> bool {
+        if self.streams.contains(stream_name) {
+            return true;
+        } else {
+            return false;
+        }
+    }
+
     // validate the storage, if the proper path for staging directory is provided
     // if the proper data directory is provided, or s3 bucket is provided etc
     pub async fn validate_storage(&self) -> Result<Option<Bytes>, ObjectStorageError> {