fixes for quest OSS

Author: parmesant

src/alerts/alert_structs.rs

@@ -713,7 +713,7 @@ pub struct DailyMTTRStats {
 pub struct MTTRHistory {
     /// Array of daily MTTR statistics
     pub daily_stats: Vec<DailyMTTRStats>,
-    pub tenant_id: Option<String>
+    pub tenant_id: Option<String>,
 }
 
 /// Query parameters for MTTR API endpoint

src/alerts/alert_traits.rs

@@ -87,27 +87,38 @@ pub trait AlertManagerTrait: Send + Sync {
         session: SessionKey,
         tags: Vec<String>,
     ) -> Result<Vec<AlertConfig>, AlertError>;
-    async fn get_alert_by_id(&self, id: Ulid, tenant_id: &Option<String>) -> Result<Box<dyn AlertTrait>, AlertError>;
+    async fn get_alert_by_id(
+        &self,
+        id: Ulid,
+        tenant_id: &Option<String>,
+    ) -> Result<Box<dyn AlertTrait>, AlertError>;
     async fn update(&self, alert: &dyn AlertTrait);
     async fn update_state(
         &self,
         alert_id: Ulid,
         new_state: AlertState,
         trigger_notif: Option<String>,
-        tenant_id: &Option<String>
+        tenant_id: &Option<String>,
     ) -> Result<(), AlertError>;
     async fn update_notification_state(
         &self,
         alert_id: Ulid,
         new_notification_state: NotificationState,
-        tenant_id: &Option<String>
+        tenant_id: &Option<String>,
     ) -> Result<(), AlertError>;
     async fn delete(&self, alert_id: Ulid, tenant_id: &Option<String>) -> Result<(), AlertError>;
-    async fn get_state(&self, alert_id: Ulid, tenant_id: &Option<String>) -> Result<AlertState, AlertError>;
+    async fn get_state(
+        &self,
+        alert_id: Ulid,
+        tenant_id: &Option<String>,
+    ) -> Result<AlertState, AlertError>;
     async fn start_task(&self, alert: Box<dyn AlertTrait>) -> Result<(), AlertError>;
     async fn delete_task(&self, alert_id: Ulid) -> Result<(), AlertError>;
     async fn list_tags(&self, tenant_id: &Option<String>) -> Vec<String>;
-    async fn get_all_alerts(&self, tenant_id: &Option<String>) -> HashMap<Ulid, Box<dyn AlertTrait>>;
+    async fn get_all_alerts(
+        &self,
+        tenant_id: &Option<String>,
+    ) -> HashMap<Ulid, Box<dyn AlertTrait>>;
 }
 
 #[async_trait]

src/alerts/alert_types.rs

@@ -110,11 +110,11 @@ impl AlertTrait for ThresholdAlert {
                     // save that user's basic auth
                     // use that to send request
                     None
-                },
+                }
                 crate::rbac::user::UserType::OAuth(_) => {
                     tracing::warn!("admin user is oauth");
                     None
-                },
+                }
             }
         } else {
             None

src/alerts/alerts_utils.rs

@@ -82,7 +82,7 @@ pub async fn execute_alert_query(
 ) -> Result<AlertQueryResult, AlertError> {
     match PARSEABLE.options.mode {
         Mode::All | Mode::Query => execute_local_query(query, time_range, tenant_id).await,
-        Mode::Prism => execute_remote_query(auth_token, query, time_range).await,
+        Mode::Prism => execute_remote_query(auth_token, query, time_range, tenant_id).await,
         _ => Err(AlertError::CustomError(format!(
             "Unsupported mode '{:?}' for alert evaluation",
             PARSEABLE.options.mode
@@ -131,6 +131,7 @@ async fn execute_remote_query(
     auth_token: Option<String>,
     query: &str,
     time_range: &TimeRange,
+    tenant_id: &Option<String>,
 ) -> Result<AlertQueryResult, AlertError> {
     let session_state = QUERY_SESSION.get_ctx().state();
     let raw_logical_plan = session_state.create_logical_plan(query).await?;
@@ -145,7 +146,7 @@ async fn execute_remote_query(
         filter_tags: None,
     };
 
-    let (result_value, _) = send_query_request(None,&query_request)
+    let (result_value, _) = send_query_request(None, &query_request, tenant_id)
         .await
         .map_err(|err| AlertError::CustomError(format!("Failed to send query request: {err}")))?;
 

src/alerts/mod.rs

@@ -1043,7 +1043,11 @@ impl AlertManagerTrait for Alerts {
         let mut map = self.alerts.write().await;
 
         for (tenant_id, raw_bytes) in raw_objects {
-            let tenant = &Some(tenant_id.clone());
+            let tenant = if tenant_id.is_empty() {
+                &None
+            } else {
+                &Some(tenant_id.clone())
+            };
             for alert_bytes in raw_bytes {
                 // First, try to parse as JSON Value to check version
                 let json_value: JsonValue = match serde_json::from_slice(&alert_bytes) {
@@ -1233,25 +1237,17 @@ impl AlertManagerTrait for Alerts {
                 "No alert found for the given ID- {id}"
             )))
         }
-        // if let Some(alert) = read_access.get(&id) {
-        //     Ok(alert.clone_box())
-        // } else {
-        //     Err(AlertError::CustomError(format!(
-        //         "No alert found for the given ID- {id}"
-        //     )))
-        // }
     }
 
     /// 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);
-        if let Some(alerts) = self.alerts.write().await.get_mut(tenant) {
-            alerts.insert(*alert.get_id(), alert.clone_box());
-        }
-        // self.alerts
-        //     .write()
-        //     .await
-        //     .insert(*alert.get_id(), alert.clone_box());
+        self.alerts
+            .write()
+            .await
+            .entry(tenant.to_owned())
+            .or_default()
+            .insert(*alert.get_id(), alert.clone_box());
     }
 
     /// Update the state of alert

src/alerts/target.rs

@@ -341,7 +341,8 @@ impl Target {
 
             match retry {
                 Retry::Infinite => loop {
-                    let current_state = if let Ok(state) = alerts.get_state(alert_id, &tenant_id).await
+                    let current_state = if let Ok(state) =
+                        alerts.get_state(alert_id, &tenant_id).await
                     {
                         state
                     } else {

src/analytics.rs

@@ -119,7 +119,8 @@ impl Report {
 
         // check liveness of indexers
         // get the count of active and inactive indexers
-        let indexer_infos: Vec<NodeMetadata> = cluster::get_node_info(NodeType::Indexer).await?;
+        let indexer_infos: Vec<NodeMetadata> =
+            cluster::get_node_info(NodeType::Indexer, &None).await?;
         for indexer in indexer_infos {
             if check_liveness(&indexer.domain_name).await {
                 active_indexers += 1;
@@ -130,7 +131,8 @@ impl Report {
 
         // check liveness of queriers
         // get the count of active and inactive queriers
-        let query_infos: Vec<NodeMetadata> = cluster::get_node_info(NodeType::Querier).await?;
+        let query_infos: Vec<NodeMetadata> =
+            cluster::get_node_info(NodeType::Querier, &None).await?;
         for query in query_infos {
             if check_liveness(&query.domain_name).await {
                 active_queriers += 1;
@@ -274,7 +276,8 @@ async fn fetch_ingestors_metrics()
         // send analytics for ingest servers
 
         // ingestor infos should be valid here, if not some thing is wrong
-        let ingestor_infos: Vec<NodeMetadata> = cluster::get_node_info(NodeType::Ingestor).await?;
+        let ingestor_infos: Vec<NodeMetadata> =
+            cluster::get_node_info(NodeType::Ingestor, &None).await?;
 
         for im in ingestor_infos {
             if !check_liveness(&im.domain_name).await {

src/banner.rs

@@ -95,7 +95,7 @@ fn status_info(config: &Parseable, scheme: &str, id: Uid) {
 /// - Store (path where the data is stored and its latency)
 async fn storage_info(config: &Parseable) {
     let storage = config.storage();
-    let latency = storage.get_object_store().get_latency().await;
+    let latency = storage.get_object_store().get_latency(&None).await;
 
     eprintln!(
         "

src/catalog/mod.rs

@@ -35,7 +35,7 @@ use crate::{
     },
     metrics::{EVENTS_INGESTED_DATE, EVENTS_INGESTED_SIZE_DATE, EVENTS_STORAGE_SIZE_DATE},
     option::Mode,
-    parseable::PARSEABLE,
+    parseable::{DEFAULT_TENANT, PARSEABLE},
     query::PartialTimeFilter,
     stats::{event_labels_date, get_current_stats, storage_size_labels_date, update_deleted_stats},
     storage::{
@@ -187,10 +187,14 @@ fn create_partition_bounds(lower_bound: DateTime<Utc>) -> (DateTime<Utc>, DateTi
 }
 
 /// Extracts statistics from live metrics for a given partition date
-fn extract_partition_metrics(stream_name: &str, partition_lower: DateTime<Utc>) -> (u64, u64, u64) {
+fn extract_partition_metrics(
+    stream_name: &str,
+    partition_lower: DateTime<Utc>,
+    tenant_id: &str,
+) -> (u64, u64, u64) {
     let date_str = partition_lower.date_naive().to_string();
-    let event_labels = event_labels_date(stream_name, "json", &date_str);
-    let storage_labels = storage_size_labels_date(stream_name, &date_str);
+    let event_labels = event_labels_date(stream_name, "json", &date_str, tenant_id);
+    let storage_labels = storage_size_labels_date(stream_name, &date_str, tenant_id);
 
     let events_ingested = EVENTS_INGESTED_DATE
         .get_metric_with_label_values(&event_labels)
@@ -220,8 +224,11 @@ async fn process_partition_groups(
     let mut new_manifest_entries = Vec::new();
 
     for ((partition_lower, _partition_upper), partition_changes) in partition_groups {
-        let (events_ingested, ingestion_size, storage_size) =
-            extract_partition_metrics(stream_name, partition_lower);
+        let (events_ingested, ingestion_size, storage_size) = extract_partition_metrics(
+            stream_name,
+            partition_lower,
+            tenant_id.as_ref().map_or(DEFAULT_TENANT, |v| v),
+        );
 
         let manifest_entry = process_single_partition(
             partition_lower,
@@ -530,7 +537,7 @@ pub async fn remove_manifest_from_snapshot(
         let stream_name_clone = stream_name.to_string();
         let dates_clone = dates.clone();
 
-        for_each_live_node(move |ingestor| {
+        for_each_live_node(tenant_id, move |ingestor| {
             let stream_name = stream_name_clone.clone();
             let dates = dates_clone.clone();
             async move {

src/event/format/json.rs

@@ -157,7 +157,7 @@ impl EventFormat for Event {
         stream_type: StreamType,
         p_custom_fields: &HashMap<String, String>,
         telemetry_type: TelemetryType,
-        tenant_id: &Option<String>
+        tenant_id: &Option<String>,
     ) -> Result<super::Event, anyhow::Error> {
         let custom_partition_values = match custom_partitions.as_ref() {
             Some(custom_partition) => {
@@ -191,7 +191,7 @@ impl EventFormat for Event {
             custom_partition_values,
             stream_type,
             telemetry_type,
-            tenant_id: tenant_id.to_owned()
+            tenant_id: tenant_id.to_owned(),
         })
     }
 }