coderabbit suggestions

Author: parmesant

src/connectors/kafka/processor.rs

@@ -64,6 +64,7 @@ impl ParseableSinkProcessor {
                 vec![log_source_entry],
                 TelemetryType::default(),
                 tenant_id,
+                None,
             )
             .await?;
 

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)
     }

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;

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() {
@@ -205,12 +205,6 @@ where
                         t = 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))

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);
             }
         }
 

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,

src/prism/home/mod.rs

@@ -1,3 +1,5 @@
+use std::collections::HashMap;
+
 /*
  * Parseable Server (C) 2022 - 2025 Parseable, Inc.
  *
@@ -27,7 +29,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 +157,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))
+        .unwrap_or(&HashMap::default())
+        .len();
     let user_added = user_count > 1; // more than just the default admin user
 
     // Calculate triggered alerts count

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();

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,

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())
+            };
         }
     }