coderabbit suggestions

parmesant · parmesant · commit e8a40fd75110 · 2026-02-13T15:33:12.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,7 +695,19 @@ 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);
                                 let user_id = filter_value
@@ -933,7 +950,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();
@@ -373,6 +377,8 @@ impl UserGroup {
                     }
                 }
             }
+        } else {
+            non_existent_users.extend(self.users.iter().map(|u| u.userid().to_string()));
         }
 
         if !non_existent_roles.is_empty() || !non_existent_users.is_empty() || !valid_name {
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
@@ -127,10 +127,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>) {

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();`
`@@ -373,6 +377,8 @@ impl UserGroup {`
`373`	`377`	`}`
`374`	`378`	`}`
`375`	`379`	`}`
	`380`	`+ } else {`
	`381`	`+ non_existent_users.extend(self.users.iter().map(\|u\| u.userid().to_string()));`
`376`	`382`	`}`
`377`	`383`
`378`	`384`	`if !non_existent_roles.is_empty() \|\| !non_existent_users.is_empty() \|\| !valid_name {`
Original file line number	Diff line number	Diff line change
`@@ -70,12 +70,18 @@ impl TenantMetadata {`
`70`	`70`	`pub fn suspend_service(&self, tenant_id: &str, service: Service) {`
`71`	`71`	`if let Some(mut tenant) = self.tenants.get_mut(tenant_id) {`
`72`	`72`	`tenant.suspended_services.insert(service);`
	`73`	`+ tenant.meta.suspended_services = Some(tenant.suspended_services.clone());`
`73`	`74`	`}`
`74`	`75`	`}`
`75`	`76`
`76`	`77`	`pub fn resume_service(&self, tenant_id: &str, service: Service) {`
`77`	`78`	`if let Some(mut tenant) = self.tenants.get_mut(tenant_id) {`
`78`	`79`	`tenant.suspended_services.remove(&service);`
	`80`	`+ tenant.meta.suspended_services = if tenant.suspended_services.is_empty() {`
	`81`	`+ None`
	`82`	`+ } else {`
	`83`	`+ Some(tenant.suspended_services.clone())`
	`84`	`+ };`
`79`	`85`	`}`
`80`	`86`	`}`
`81`	`87`