lint related changes

Author: parmesant

src/alerts/alert_structs.rs

@@ -75,9 +75,11 @@ pub struct BasicAlertFields {
     pub severity: Severity,
 }
 
+pub type AlertMap = HashMap<Ulid, Box<dyn AlertTrait>>;
+
 #[derive(Debug)]
 pub struct Alerts {
-    pub alerts: RwLock<HashMap<String, HashMap<Ulid, Box<dyn AlertTrait>>>>,
+    pub alerts: RwLock<HashMap<String, AlertMap>>,
     pub sender: mpsc::Sender<AlertTask>,
 }
 

src/alerts/alert_types.rs

@@ -18,6 +18,7 @@
 
 use std::{str::FromStr, time::Duration};
 
+use actix_web::http::header::{HeaderMap, HeaderName, HeaderValue};
 use chrono::{DateTime, Utc};
 use serde_json::Value;
 use tonic::async_trait;
@@ -39,11 +40,9 @@ use crate::{
     metastore::metastore_traits::MetastoreObject,
     parseable::PARSEABLE,
     query::resolve_stream_names,
-    rbac::{
-        map::{SessionKey, roles, users},
-        role::model::DefaultPrivilege,
-    },
+    rbac::map::SessionKey,
     storage::object_storage::alert_json_path,
+    tenants::TENANT_METADATA,
     utils::user_auth_for_query,
 };
 
@@ -88,34 +87,15 @@ impl MetastoreObject for ThresholdAlert {
 impl AlertTrait for ThresholdAlert {
     async fn eval_alert(&self) -> Result<Option<String>, AlertError> {
         let time_range = extract_time_range(&self.eval_config)?;
-        let auth = if let Some(tenant) = &self.tenant_id
-            && let Some(tenant_users) = users().get(tenant)
-            && let Some(tenant_roles) = roles().get(tenant)
-            && let Some(user) = tenant_users.iter().find_map(|(_, user)| {
-                let mut res = None;
-                for role in &user.roles {
-                    if let Some(role) = tenant_roles.get(role)
-                        && role.contains(&DefaultPrivilege::Admin)
-                    {
-                        res = Some(user.clone());
-                        break;
-                    }
-                }
-                res
-            }) {
-            // fetch admin credentials for tenant
-            match user.ty {
-                crate::rbac::user::UserType::Native(basic) => {
-                    // Create a protected user whose details can't be edited
-                    // save that user's basic auth
-                    // use that to send request
-                    None
-                }
-                crate::rbac::user::UserType::OAuth(_) => {
-                    tracing::warn!("admin user is oauth");
-                    None
-                }
-            }
+        let auth = if let Some(tenant) = self.tenant_id.as_ref()
+            && let Some(header) = TENANT_METADATA.get_global_query_auth(tenant)
+        {
+            let mut map = HeaderMap::new();
+            map.insert(
+                HeaderName::from_static("authorization"),
+                HeaderValue::from_str(&header).unwrap(),
+            );
+            Some(map)
         } else {
             None
         };

src/alerts/alerts_utils.rs

@@ -18,7 +18,7 @@
 
 use std::{collections::HashMap, fmt::Display};
 
-use actix_web::Either;
+use actix_web::{Either, http::header::HeaderMap};
 use arrow_array::{Array, Float64Array, Int64Array, RecordBatch};
 use datafusion::{
     logical_expr::{Literal, LogicalPlan},
@@ -75,7 +75,7 @@ pub fn extract_time_range(eval_config: &super::EvalConfig) -> Result<TimeRange,
 
 /// Execute the alert query based on the current mode and return structured group results
 pub async fn execute_alert_query(
-    auth_token: Option<String>,
+    auth_token: Option<HeaderMap>,
     query: &str,
     time_range: &TimeRange,
     tenant_id: &Option<String>,
@@ -128,7 +128,7 @@ async fn execute_local_query(
 
 /// Execute alert query remotely (Prism mode)
 async fn execute_remote_query(
-    auth_token: Option<String>,
+    auth_token: Option<HeaderMap>,
     query: &str,
     time_range: &TimeRange,
     tenant_id: &Option<String>,
@@ -146,7 +146,7 @@ async fn execute_remote_query(
         filter_tags: None,
     };
 
-    let (result_value, _) = send_query_request(None, &query_request, tenant_id)
+    let (result_value, _) = send_query_request(auth_token, &query_request, tenant_id)
         .await
         .map_err(|err| AlertError::CustomError(format!("Failed to send query request: {err}")))?;
 

src/alerts/mod.rs

@@ -624,9 +624,9 @@ impl AlertConfig {
         }
 
         // get active sessions
-        let active_sessions = sessions().get_active_sessions();
+        let active_session = sessions().get_active_sessions();
         let mut broadcast_to = vec![];
-        for session in active_sessions {
+        for (session, _, _) in active_session {
             if user_auth_for_query(&session, &self.query).await.is_ok()
                 && let SessionKey::SessionId(id) = &session
             {
@@ -1150,7 +1150,7 @@ impl AlertManagerTrait for Alerts {
         tags: Vec<String>,
     ) -> Result<Vec<AlertConfig>, AlertError> {
         let tenant_id = get_tenant_id_from_key(&session);
-        let tenant = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         // First, collect all alerts without performing auth checks to avoid holding the lock
         let all_alerts: Vec<AlertConfig> = {
             let alerts_guard = self.alerts.read().await;
@@ -1226,7 +1226,7 @@ impl AlertManagerTrait for Alerts {
         id: Ulid,
         tenant_id: &Option<String>,
     ) -> Result<Box<dyn AlertTrait>, AlertError> {
-        let tenant = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         let read_access = self.alerts.read().await;
         if let Some(alerts) = read_access.get(tenant)
             && let Some(alert) = alerts.get(&id)
@@ -1241,7 +1241,7 @@ impl AlertManagerTrait for Alerts {
 
     /// Update the in-mem vector of alerts
     async fn update(&self, alert: &dyn AlertTrait) {
-        let tenant = alert.get_tenant_id().as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant = alert.get_tenant_id().as_deref().unwrap_or(DEFAULT_TENANT);
         self.alerts
             .write()
             .await
@@ -1331,7 +1331,7 @@ impl AlertManagerTrait for Alerts {
         {
             let mut write_access = self.alerts.write().await;
 
-            let tenant = alert.get_tenant_id().as_ref().map_or(DEFAULT_TENANT, |v| v);
+            let tenant = alert.get_tenant_id().as_deref().unwrap_or(DEFAULT_TENANT);
             if let Some(alerts) = write_access.get_mut(tenant) {
                 alerts.insert(*alert.get_id(), alert.clone_box());
             }
@@ -1350,7 +1350,7 @@ impl AlertManagerTrait for Alerts {
     ) -> Result<(), AlertError> {
         // read and modify alert
         let mut write_access = self.alerts.write().await;
-        let tenant = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         let mut alert: Box<dyn AlertTrait> = if let Some(alerts) = write_access.get(tenant)
             && let Some(alert) = alerts.get(&alert_id)
         {
@@ -1384,7 +1384,7 @@ impl AlertManagerTrait for Alerts {
 
     /// Remove alert and scheduled task from disk and memory
     async fn delete(&self, alert_id: Ulid, tenant_id: &Option<String>) -> Result<(), AlertError> {
-        let tenant = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         if let Some(alerts) = self.alerts.write().await.get_mut(tenant)
             && let Some(_) = alerts.remove(&alert_id)
         {
@@ -1406,7 +1406,7 @@ impl AlertManagerTrait for Alerts {
         alert_id: Ulid,
         tenant_id: &Option<String>,
     ) -> Result<AlertState, AlertError> {
-        let tenant = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         let read_access = self.alerts.read().await;
 
         if let Some(alerts) = read_access.get(tenant)
@@ -1441,7 +1441,7 @@ impl AlertManagerTrait for Alerts {
     /// List tags from all alerts
     /// This function returns a list of unique tags from all alerts
     async fn list_tags(&self, tenant_id: &Option<String>) -> Vec<String> {
-        let tenant = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         // let alerts = self.alerts.read().await;
         let mut tags = if let Some(alerts) = self.alerts.read().await.get(tenant) {
             alerts
@@ -1461,7 +1461,7 @@ impl AlertManagerTrait for Alerts {
         &self,
         tenant_id: &Option<String>,
     ) -> HashMap<Ulid, Box<dyn AlertTrait>> {
-        let tenant = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         if let Some(alerts) = self.alerts.read().await.get(tenant) {
             alerts.iter().map(|(k, v)| (*k, v.clone_box())).collect()
         } else {

src/alerts/target.rs

@@ -78,7 +78,7 @@ impl TargetConfigs {
             .put_target(&target, &target.tenant)
             .await?;
         let mut map = self.target_configs.write().await;
-        let tenant_id = target.tenant.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT);
         map.entry(tenant_id.to_owned())
             .or_default()
             .insert(target.id, target);
@@ -87,7 +87,7 @@ impl TargetConfigs {
     }
 
     pub async fn list(&self, tenant_id: &Option<String>) -> Result<Vec<Target>, AlertError> {
-        let tenant_id = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         let targets = if let Some(targets) = self.target_configs.read().await.get(tenant_id) {
             targets.values().cloned().collect_vec()
         } else {
@@ -108,7 +108,7 @@ impl TargetConfigs {
         target_id: &Ulid,
         tenant_id: &Option<String>,
     ) -> Result<Target, AlertError> {
-        let tenant_id = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         let target = if let Some(targets) = self.target_configs.read().await.get(tenant_id) {
             targets
                 .get(target_id)
@@ -133,7 +133,7 @@ impl TargetConfigs {
         target_id: &Ulid,
         tenant_id: &Option<String>,
     ) -> Result<Target, AlertError> {
-        let tenant = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         // ensure that the target is not being used by any alert
         let guard = ALERTS.read().await;
         let alerts = if let Some(alerts) = guard.as_ref() {

src/analytics.rs

@@ -207,7 +207,7 @@ fn total_event_stats() -> (Stats, Stats, Stats) {
     let mut deleted_json_bytes: u64 = 0;
 
     let tenants = if let Some(tenants) = PARSEABLE.list_tenants() {
-        tenants.into_iter().map(|v| Some(v)).collect()
+        tenants.into_iter().map(Some).collect()
     } else {
         vec![None]
     };

src/catalog/mod.rs

@@ -464,7 +464,7 @@ async fn create_manifest(
         .get_manifest_path(stream_name, lower_bound, upper_bound, tenant_id)
         .await
         .map_err(|e| ObjectStorageError::MetastoreError(Box::new(e.to_detail())))?;
-    tracing::warn!("manifest path_url= {path_url}");
+
     let new_snapshot_entry = snapshot::ManifestItem {
         manifest_path: path_url.to_owned(),
         time_lower_bound: lower_bound,

src/correlation.rs

@@ -84,7 +84,7 @@ impl Correlations {
         let mut user_correlations = vec![];
         let permissions = Users.get_permissions(session_key);
         let tenant_id = get_tenant_id_from_key(session_key);
-        let tenant = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         if let Some(corrs) = self.read().await.get(tenant) {
             for correlation in corrs.values() {
                 let tables = &correlation
@@ -109,17 +109,17 @@ impl Correlations {
         correlation_id: &str,
         tenant_id: &Option<String>,
     ) -> Result<CorrelationConfig, CorrelationError> {
-        let tenant_id = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         if let Some(corrs) = self.read().await.get(tenant_id) {
             corrs.get(correlation_id).cloned().ok_or_else(|| {
                 CorrelationError::AnyhowError(anyhow::Error::msg(format!(
                     "Unable to find correlation with ID- {correlation_id}"
                 )))
             })
         } else {
-            return Err(CorrelationError::AnyhowError(anyhow::Error::msg(format!(
+            Err(CorrelationError::AnyhowError(anyhow::Error::msg(format!(
                 "Unable to find correlation with ID- {correlation_id}"
-            ))));
+            ))))
         }
     }
 
@@ -137,7 +137,7 @@ impl Correlations {
             .metastore
             .put_correlation(&correlation, &tenant_id)
             .await?;
-        let tenant = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        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());
@@ -173,7 +173,7 @@ impl Correlations {
             .put_correlation(&updated_correlation, &tenant_id)
             .await?;
 
-        let tenant = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        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(

src/event/mod.rs

@@ -87,7 +87,7 @@ impl Event {
                 self.stream_type,
             )?;
 
-        let tenant = self.tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+        let tenant = self.tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         update_stats(
             &self.stream_name,
             self.origin_format,
@@ -145,7 +145,7 @@ pub fn commit_schema(
     tenant_id: &Option<String>,
 ) -> Result<(), StagingError> {
     let mut stream_metadata = PARSEABLE.streams.write().expect("lock poisoned");
-    let tenant_id = tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v);
+    let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
     let map = &mut stream_metadata
         .get_mut(tenant_id)
         .ok_or_else(|| TenantNotFound(tenant_id.to_owned()))?

src/handlers/http/cluster/mod.rs

@@ -45,7 +45,7 @@ use crate::handlers::http::modal::ingest::SyncRole;
 use crate::handlers::http::query::{Query, QueryError, TIME_ELAPSED_HEADER};
 use crate::metrics::prom_utils::Metrics;
 use crate::option::Mode;
-use crate::parseable::{DEFAULT_TENANT, PARSEABLE};
+use crate::parseable::PARSEABLE;
 use crate::rbac::role::model::DefaultPrivilege;
 use crate::rbac::user::User;
 use crate::stats::Stats;
@@ -710,7 +710,6 @@ pub async fn sync_password_reset_with_ingestors(
 
 // forward the put role request to all ingestors and queriers to keep them in sync
 pub async fn sync_role_update(
-    req: HttpRequest,
     name: String,
     privileges: Vec<DefaultPrivilege>,
     tenant_id: &Option<String>,
@@ -1902,6 +1901,7 @@ pub async fn send_query_request(
     let mut map = reqwest::header::HeaderMap::new();
 
     if let Some(auth) = auth_token {
+        // always basic auth
         for (key, value) in auth.iter() {
             if let Ok(name) = reqwest::header::HeaderName::from_bytes(key.as_str().as_bytes())
                 && let Ok(val) = reqwest::header::HeaderValue::from_bytes(value.as_bytes())