coderabbit suggestions

parmesant · parmesant · commit 3186ae1d6176 · 2026-02-13T16:17:33.000+05:30
diff --git a/src/connectors/kafka/processor.rs b/src/connectors/kafka/processor.rs
@@ -64,6 +64,7 @@ impl ParseableSinkProcessor {
                 vec![log_source_entry],
                 TelemetryType::default(),
                 tenant_id,
+                None,
             )
             .await?;
 
diff --git a/src/correlation.rs b/src/correlation.rs
@@ -139,9 +139,11 @@ impl Correlations {
             .await?;
         let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         // Update in memory
-        if let Some(corrs) = self.write().await.get_mut(tenant) {
-            corrs.insert(correlation.id.to_owned(), correlation.clone());
-        }
+        self.write()
+            .await
+            .entry(tenant.to_string())
+            .or_default()
+            .insert(correlation.id.to_owned(), correlation.clone());
 
         Ok(correlation)
     }
@@ -208,7 +210,11 @@ impl Correlations {
             .await?;
 
         // Delete from memory
-        self.write().await.remove(&correlation.id);
+        self.write()
+            .await
+            .entry(tenant_id.as_deref().unwrap_or(DEFAULT_TENANT).to_owned())
+            .or_default()
+            .remove(&correlation.id);
 
         Ok(())
     }
diff --git a/src/handlers/http/ingest.rs b/src/handlers/http/ingest.rs
@@ -120,7 +120,7 @@ pub async fn ingest(
             vec![log_source_entry.clone()],
             telemetry_type,
             &tenant_id,
-            None
+            None,
         )
         .await
         .map_err(|e| {
@@ -239,7 +239,7 @@ pub async fn setup_otel_stream(
             vec![log_source_entry.clone()],
             telemetry_type,
             &tenant_id,
-            None
+            None,
         )
         .await?;
     let mut time_partition = None;
diff --git a/src/handlers/http/middleware.rs b/src/handlers/http/middleware.rs
@@ -168,7 +168,7 @@ where
         // if action is Ingest and multi-tenancy is on, then request MUST have tenant id
         // else check for the presence of tenant id using other details
 
-        // an optional error to track the presence of CORRECTtenant header in case of ingestion
+        // an optional error to track the presence of CORRECT tenant header in case of ingestion
         let mut header_error = None;
         let user_and_tenant_id: Result<(Result<String, RBACError>, Option<String>), RBACError> =
             if PARSEABLE.options.is_multi_tenant() {
@@ -203,14 +203,11 @@ where
                             HeaderValue::from_str(tid).unwrap(),
                         );
                         t = tenant;
+                    } else {
+                        // remove the header if already present
+                        req.headers_mut().remove("tenant");
                     }
                     t
-                    // else {
-                    //     header_error = Some(actix_web::Error::from(PostError::Header(
-                    //         crate::utils::header_parsing::ParseHeaderError::InvalidTenantId,
-                    //     )));
-                    //     None
-                    // }
                 };
                 let userid = get_user_from_request(req.request());
                 Ok((userid, tenant))
diff --git a/src/metastore/metastores/object_store_metastore.rs b/src/metastore/metastores/object_store_metastore.rs
@@ -514,9 +514,9 @@ impl Metastore for ObjectStoreMetastore {
 
     /// Fetch all dashboards
     async fn get_dashboards(&self) -> Result<HashMap<String, Vec<Bytes>>, MetastoreError> {
-        let mut dashboards = HashMap::new();
+        let mut dashboards: HashMap<String, Vec<Bytes>> = HashMap::new();
         let base_paths = PARSEABLE.list_tenants().unwrap_or_else(|| vec!["".into()]);
-        for mut tenant in base_paths {
+        for tenant in base_paths {
             let tenant_id = &Some(tenant.clone());
             let users_dir = RelativePathBuf::from_iter([&tenant, USERS_ROOT_DIR]);
             for user in self
@@ -533,11 +533,16 @@ impl Metastore for ObjectStoreMetastore {
                         tenant_id,
                     )
                     .await?;
-                if tenant.is_empty() {
-                    tenant.clone_from(&DEFAULT_TENANT.to_string());
-                }
-                dashboards.insert(tenant.to_owned(), dashboard_bytes);
-                // dashboards.extend(dashboard_bytes);
+
+                let tenant_key = if tenant.is_empty() {
+                    DEFAULT_TENANT.to_string()
+                } else {
+                    tenant.clone()
+                };
+                dashboards
+                    .entry(tenant_key)
+                    .or_default()
+                    .extend(dashboard_bytes);
             }
         }
 
@@ -690,9 +695,21 @@ impl Metastore for ObjectStoreMetastore {
 
                             if version == Some("v1") {
                                 // delete older version of the filter
-                                self.storage.delete_object(&filters_path, tenant_id).await?;
+                                // get filter id to delete
+                                let filterid = meta
+                                    .get("filter_id")
+                                    .and_then(|filter_id| filter_id.as_str());
+                                if let Some(filterid) = filterid {
+                                    self.storage
+                                        .delete_object(
+                                            &filters_path.join(format!("{filterid}.json")),
+                                            tenant_id,
+                                        )
+                                        .await?;
+                                }
+                                // self.storage.delete_object(&filters_path, tenant_id).await?;
 
-                                filter_value = migrate_v1_v2(filter_value);
+                                filter_value = migrate_v1_v2(filter_value, tenant_id);
                                 let user_id = filter_value
                                     .as_object()
                                     .unwrap()
@@ -717,6 +734,7 @@ impl Metastore for ObjectStoreMetastore {
                                         user_id,
                                         stream_name,
                                         &format!("{filter_id}.json"),
+                                        tenant_id,
                                     );
                                     let filter_bytes = to_bytes(&filter_value);
                                     self.storage
@@ -933,7 +951,11 @@ impl Metastore for ObjectStoreMetastore {
             .collect::<Vec<_>>();
 
         for date in dates {
-            let date_path = object_store::path::Path::from(format!("{}/{}", stream_name, &date));
+            let date_path = if let Some(tenant) = tenant_id {
+                object_store::path::Path::from(format!("{}/{}/{}", tenant, stream_name, &date))
+            } else {
+                object_store::path::Path::from(format!("{}/{}", stream_name, &date))
+            };
             let resp = self.storage.list_with_delimiter(Some(date_path)).await?;
 
             let manifest_paths: Vec<String> = resp
diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs
@@ -473,7 +473,7 @@ impl Parseable {
                     vec![log_source_entry.clone()],
                     TelemetryType::Logs,
                     &tenant_id,
-                    None
+                    None,
                 )
                 .await;
 
@@ -512,6 +512,7 @@ impl Parseable {
     }
 
     // Check if the stream exists and create a new stream if doesn't exist
+    #[allow(clippy::too_many_arguments)]
     pub async fn create_stream_if_not_exists(
         &self,
         stream_name: &str,
@@ -1045,14 +1046,15 @@ impl Parseable {
             return Err(anyhow::Error::msg("P_MULTI_TENANCY is set to false"));
         }
 
-        if self.tenants.read().unwrap().contains(&tenant_id) {
+        let mut tenants = self.tenants.write().unwrap();
+        if tenants.contains(&tenant_id) {
             return Err(anyhow::Error::msg(format!(
                 "Tenant with id- {tenant_id} already exists"
             )));
-        } else {
-            self.tenants.write().unwrap().push(tenant_id.clone());
-            TENANT_METADATA.insert_tenant(tenant_id, tenant_meta);
         }
+        tenants.push(tenant_id.clone());
+        drop(tenants);
+        TENANT_METADATA.insert_tenant(tenant_id, tenant_meta);
 
         Ok(())
     }
@@ -1120,6 +1122,9 @@ impl Parseable {
         // delete resources
 
         // delete from in-mem
+        if let Ok(mut tenants) = self.tenants.write() {
+            tenants.retain(|t| t != tenant_id);
+        }
         TENANT_METADATA.delete_tenant(tenant_id);
         Ok(())
     }
@@ -1154,7 +1159,9 @@ impl Parseable {
             }
         }
 
-        if let Ok(mut t) = self.tenants.write() {
+        if let Ok(mut t) = self.tenants.write()
+            && is_multi_tenant
+        {
             t.extend(dirs);
             Ok(Some(()))
         } else {
diff --git a/src/prism/home/mod.rs b/src/prism/home/mod.rs
@@ -27,7 +27,7 @@ use crate::{
     event::format::{LogSource, LogSourceEntry},
     handlers::{TelemetryType, http::logstream::error::StreamError},
     metastore::MetastoreError,
-    parseable::PARSEABLE,
+    parseable::{DEFAULT_TENANT, PARSEABLE},
     rbac::{
         Users,
         map::{SessionKey, users},
@@ -155,7 +155,10 @@ pub async fn generate_home_response(
 
     // Generate checklist and count triggered alerts
     let data_ingested = datasets.iter().any(|d| d.ingestion);
-    let user_count = users().len();
+    let user_count = users()
+        .get(tenant_id.as_deref().unwrap_or(DEFAULT_TENANT))
+        .map(|m| m.len())
+        .unwrap_or(0);
     let user_added = user_count > 1; // more than just the default admin user
 
     // Calculate triggered alerts count
diff --git a/src/rbac/user.rs b/src/rbac/user.rs
@@ -349,7 +349,9 @@ impl UserGroup {
     pub fn validate(&self, tenant_id: &Option<String>) -> Result<(), RBACError> {
         let valid_name = is_valid_group_name(&self.name);
         let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
-        if read_user_groups().contains_key(&self.name) {
+        if let Some(tenant_ug) = read_user_groups().get(tenant)
+            && tenant_ug.contains_key(&self.name)
+        {
             return Err(RBACError::UserGroupExists(self.name.clone()));
         }
         let mut non_existent_roles = Vec::new();
@@ -361,6 +363,8 @@ impl UserGroup {
                         non_existent_roles.push(role.clone());
                     }
                 }
+            } else {
+                non_existent_roles.extend(self.roles.iter().cloned());
             }
         }
         let mut non_existent_users = Vec::new();
@@ -372,6 +376,8 @@ impl UserGroup {
                         non_existent_users.push(group_user.userid().to_string());
                     }
                 }
+            } else {
+                non_existent_users.extend(self.users.iter().map(|u| u.userid().to_string()));
             }
         }
 
@@ -496,12 +502,4 @@ impl UserGroup {
 
         self.remove_users(users_to_remove)
     }
-
-    // pub async fn update_in_metadata(&self, tenant_id: &Option<String>) -> Result<(), RBACError> {
-    //     let mut metadata = get_metadata(tenant_id).await?;
-    //     metadata.user_groups.retain(|x| x.name != self.name);
-    //     metadata.user_groups.push(self.clone());
-    //     put_metadata(&metadata).await?;
-    //     Ok(())
-    // }
 }
diff --git a/src/storage/object_storage.rs b/src/storage/object_storage.rs
@@ -1236,8 +1236,19 @@ pub fn stream_json_path(stream_name: &str, tenant_id: &Option<String>) -> Relati
 
 /// if filter_id is an empty str it should not append it to the rel path
 #[inline(always)]
-pub fn filter_path(user_id: &str, stream_name: &str, filter_file_name: &str) -> RelativePathBuf {
+pub fn filter_path(
+    user_id: &str,
+    stream_name: &str,
+    filter_file_name: &str,
+    tenant_id: &Option<String>,
+) -> RelativePathBuf {
+    let root = if let Some(tenant) = tenant_id.as_ref() {
+        tenant
+    } else {
+        ""
+    };
     RelativePathBuf::from_iter([
+        root,
         USERS_ROOT_DIR,
         user_id,
         FILTER_DIR,
diff --git a/src/storage/retention.rs b/src/storage/retention.rs
@@ -109,7 +109,13 @@ pub struct Task {
 }
 
 impl Task {
-    pub fn new(description: String, days: u32) -> Self {
+    pub fn new(description: String, mut days: u32) -> Self {
+        if days.eq(&0) {
+            days = 7;
+            tracing::warn!(
+                "Using default 7 days for retention since an invalid value of 0 was provided by the user"
+            );
+        }
         let days = NonZeroU32::new(days).unwrap();
         Self {
             description,
diff --git a/src/tenants/mod.rs b/src/tenants/mod.rs
@@ -70,12 +70,18 @@ impl TenantMetadata {
     pub fn suspend_service(&self, tenant_id: &str, service: Service) {
         if let Some(mut tenant) = self.tenants.get_mut(tenant_id) {
             tenant.suspended_services.insert(service);
+            tenant.meta.suspended_services = Some(tenant.suspended_services.clone());
         }
     }
 
     pub fn resume_service(&self, tenant_id: &str, service: Service) {
         if let Some(mut tenant) = self.tenants.get_mut(tenant_id) {
             tenant.suspended_services.remove(&service);
+            tenant.meta.suspended_services = if tenant.suspended_services.is_empty() {
+                None
+            } else {
+                Some(tenant.suspended_services.clone())
+            };
         }
     }
 
diff --git a/src/users/filters.rs b/src/users/filters.rs
@@ -43,6 +43,7 @@ pub struct Filter {
     pub filter_id: Option<String>,
     pub query: FilterQuery,
     pub time_filter: Option<TimeFilter>,
+    pub tenant_id: Option<String>,
     /// all other fields are variable and can be added as needed
     #[serde(flatten)]
     pub other_fields: Option<serde_json::Map<String, Value>>,
@@ -54,6 +55,7 @@ impl MetastoreObject for Filter {
             self.user_id.as_ref().unwrap(),
             &self.stream_name,
             &format!("{}.json", self.filter_id.as_ref().unwrap()),
+            &self.tenant_id,
         )
         .to_string()
     }
@@ -127,10 +129,10 @@ impl Filters {
 
     pub async fn update(&self, filter: &Filter, tenant_id: &Option<String>) {
         let mut s = self.0.write().await;
-        if let Some(filters) = s.get_mut(tenant_id.as_deref().unwrap_or(DEFAULT_TENANT)) {
-            filters.retain(|f| f.filter_id != filter.filter_id);
-            filters.push(filter.clone());
-        }
+        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT).to_owned();
+        let filters = s.entry(tenant).or_default();
+        filters.retain(|f| f.filter_id != filter.filter_id);
+        filters.push(filter.clone());
     }
 
     pub async fn delete_filter(&self, filter_id: &str, tenant_id: &Option<String>) {
@@ -196,7 +198,7 @@ impl Filters {
     }
 }
 
-pub fn migrate_v1_v2(mut filter_meta: Value) -> Value {
+pub fn migrate_v1_v2(mut filter_meta: Value, tenant_id: &Option<String>) -> Value {
     let filter_meta_map = filter_meta.as_object_mut().unwrap();
     let user_id = filter_meta_map.get("user_id").unwrap().clone();
     let str_user_id = user_id.as_str().unwrap();
@@ -206,6 +208,9 @@ pub fn migrate_v1_v2(mut filter_meta: Value) -> Value {
         "version".to_owned(),
         Value::String(CURRENT_FILTER_VERSION.into()),
     );
+    if let Some(tenant) = tenant_id {
+        filter_meta_map.insert("tenant_id".to_owned(), Value::String(tenant.to_owned()));
+    }
 
     filter_meta
 }

Original file line number	Diff line number	Diff line change
`@@ -64,6 +64,7 @@ impl ParseableSinkProcessor {`
`64`	`64`	`vec![log_source_entry],`
`65`	`65`	`TelemetryType::default(),`
`66`	`66`	`tenant_id,`
	`67`	`+ None,`
`67`	`68`	`)`
`68`	`69`	`.await?;`
`69`	`70`
Original file line number	Diff line number	Diff line change
`@@ -349,7 +349,9 @@ impl UserGroup {`
`349`	`349`	`pub fn validate(&self, tenant_id: &Option<String>) -> Result<(), RBACError> {`
`350`	`350`	`let valid_name = is_valid_group_name(&self.name);`
`351`	`351`	`let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);`
`352`		`- if read_user_groups().contains_key(&self.name) {`
	`352`	`+ if let Some(tenant_ug) = read_user_groups().get(tenant)`
	`353`	`+ && tenant_ug.contains_key(&self.name)`
	`354`	`+ {`
`353`	`355`	`return Err(RBACError::UserGroupExists(self.name.clone()));`
`354`	`356`	`}`
`355`	`357`	`let mut non_existent_roles = Vec::new();`
`@@ -361,6 +363,8 @@ impl UserGroup {`
`361`	`363`	`non_existent_roles.push(role.clone());`
`362`	`364`	`}`
`363`	`365`	`}`
	`366`	`+ } else {`
	`367`	`+ non_existent_roles.extend(self.roles.iter().cloned());`
`364`	`368`	`}`
`365`	`369`	`}`
`366`	`370`	`let mut non_existent_users = Vec::new();`
`@@ -372,6 +376,8 @@ impl UserGroup {`
`372`	`376`	`non_existent_users.push(group_user.userid().to_string());`
`373`	`377`	`}`
`374`	`378`	`}`
	`379`	`+ } else {`
	`380`	`+ non_existent_users.extend(self.users.iter().map(\|u\| u.userid().to_string()));`
`375`	`381`	`}`
`376`	`382`	`}`
`377`	`383`
`@@ -496,12 +502,4 @@ impl UserGroup {`
`496`	`502`
`497`	`503`	`self.remove_users(users_to_remove)`
`498`	`504`	`}`
`499`		`-`
`500`		`- // pub async fn update_in_metadata(&self, tenant_id: &Option<String>) -> Result<(), RBACError> {`
`501`		`- // let mut metadata = get_metadata(tenant_id).await?;`
`502`		`- // metadata.user_groups.retain(\|x\| x.name != self.name);`
`503`		`- // metadata.user_groups.push(self.clone());`
`504`		`- // put_metadata(&metadata).await?;`
`505`		`- // Ok(())`
`506`		`- // }`
`507`	`505`	`}`