add querier metadata

Author: nikhilsinhaparseable

src/analytics.rs

@@ -36,6 +36,7 @@ use crate::{
         http::{
             base_path_without_preceding_slash,
             cluster::{self, utils::check_liveness},
+            modal::NodeMetadata,
         },
         STREAM_NAME_HEADER_KEY,
     },
@@ -228,7 +229,7 @@ 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 = cluster::get_ingestor_info().await.unwrap();
+        let ingestor_infos: Vec<NodeMetadata> = cluster::get_node_info("ingestor").await.unwrap();
 
         for im in ingestor_infos {
             if !check_liveness(&im.domain_name).await {

src/catalog/mod.rs

@@ -28,7 +28,10 @@ use tracing::{error, info};
 
 use crate::{
     event::DEFAULT_TIMESTAMP_KEY,
-    handlers::{self, http::base_path_without_preceding_slash},
+    handlers::{
+        self,
+        http::{base_path_without_preceding_slash, modal::NodeMetadata},
+    },
     metrics::{EVENTS_INGESTED_DATE, EVENTS_INGESTED_SIZE_DATE, EVENTS_STORAGE_SIZE_DATE},
     option::Mode,
     parseable::PARSEABLE,
@@ -406,8 +409,8 @@ pub async fn get_first_event(
             }
         }
         Mode::Query => {
-            let ingestor_metadata =
-                handlers::http::cluster::get_ingestor_info()
+            let ingestor_metadata: Vec<NodeMetadata> =
+                handlers::http::cluster::get_node_info("ingestor")
                     .await
                     .map_err(|err| {
                         error!("Fatal: failed to get ingestor info: {:?}", err);

src/cli.rs

@@ -328,6 +328,14 @@ pub struct Options {
     )]
     pub indexer_endpoint: String,
 
+    #[arg(
+        long,
+        env = "P_QUERIER_ENDPOINT",
+        default_value = "",
+        help = "URL to connect to this specific querier. Default is the address of the server"
+    )]
+    pub querier_endpoint: String,
+
     #[command(flatten)]
     pub oidc: Option<OidcConfig>,
 
@@ -470,6 +478,20 @@ impl Options {
                 }
                 (&self.indexer_endpoint, "P_INDEXER_ENDPOINT")
             }
+            Mode::Query => {
+                if self.querier_endpoint.is_empty() {
+                    return format!(
+                        "{}://{}",
+                        self.get_scheme(),
+                        self.address
+                    )
+                    .parse::<Url>() // if the value was improperly set, this will panic before hand
+                    .unwrap_or_else(|err| {
+                        panic!("{err}, failed to parse `{}` as Url. Please set the environment variable `P_ADDR` to `<ip address>:<port>` without the scheme (e.g., 192.168.1.1:8000). Please refer to the documentation: https://logg.ing/env for more details.", self.address)
+                    });
+                }
+                (&self.querier_endpoint, "P_QUERIER_ENDPOINT")
+            }
             _ => panic!("Invalid mode"),
         };
 

src/handlers/airplane.rs

@@ -33,7 +33,8 @@ use futures_util::{Future, TryFutureExt};
 use tonic::transport::{Identity, Server, ServerTlsConfig};
 use tonic_web::GrpcWebLayer;
 
-use crate::handlers::http::cluster::get_ingestor_info;
+use crate::handlers::http::cluster::get_node_info;
+use crate::handlers::http::modal::NodeMetadata;
 use crate::handlers::http::query::{into_query, update_schema_when_distributed};
 use crate::handlers::livetail::cross_origin_config;
 use crate::metrics::QUERY_EXECUTE_TIME;
@@ -179,7 +180,7 @@ impl FlightService for AirServiceImpl {
             })
             .to_string();
 
-            let ingester_metadatas = get_ingestor_info()
+            let ingester_metadatas: Vec<NodeMetadata> = get_node_info("ingestor")
                 .await
                 .map_err(|err| Status::failed_precondition(err.to_string()))?;
             let mut minute_result: Vec<RecordBatch> = vec![];

src/handlers/http/cluster/mod.rs

@@ -54,14 +54,10 @@ use crate::HTTP_CLIENT;
 use super::base_path_without_preceding_slash;
 use super::ingest::PostError;
 use super::logstream::error::StreamError;
-use super::modal::{IndexerMetadata, IngestorMetadata, Metadata};
+use super::modal::{IndexerMetadata, IngestorMetadata, Metadata, NodeMetadata, QuerierMetadata};
 use super::rbac::RBACError;
 use super::role::RoleError;
 
-type IngestorMetadataArr = Vec<IngestorMetadata>;
-
-type IndexerMetadataArr = Vec<IndexerMetadata>;
-
 pub const INTERNAL_STREAM_NAME: &str = "pmeta";
 
 const CLUSTER_METRICS_INTERVAL_SECONDS: Interval = clokwerk::Interval::Minutes(1);
@@ -77,7 +73,7 @@ pub async fn sync_streams_with_ingestors(
     for (key, value) in headers.iter() {
         reqwest_headers.insert(key.clone(), value.clone());
     }
-    let ingestor_infos = get_ingestor_info().await.map_err(|err| {
+    let ingestor_infos: Vec<NodeMetadata> = get_node_info("ingestor").await.map_err(|err| {
         error!("Fatal: failed to get ingestor info: {:?}", err);
         StreamError::Anyhow(err)
     })?;
@@ -125,7 +121,7 @@ pub async fn sync_users_with_roles_with_ingestors(
     username: &String,
     role: &HashSet<String>,
 ) -> Result<(), RBACError> {
-    let ingestor_infos = get_ingestor_info().await.map_err(|err| {
+    let ingestor_infos: Vec<NodeMetadata> = get_node_info("ingestor").await.map_err(|err| {
         error!("Fatal: failed to get ingestor info: {:?}", err);
         RBACError::Anyhow(err)
     })?;
@@ -175,7 +171,7 @@ pub async fn sync_users_with_roles_with_ingestors(
 
 // forward the delete user request to all ingestors to keep them in sync
 pub async fn sync_user_deletion_with_ingestors(username: &String) -> Result<(), RBACError> {
-    let ingestor_infos = get_ingestor_info().await.map_err(|err| {
+    let ingestor_infos: Vec<NodeMetadata> = get_node_info("ingestor").await.map_err(|err| {
         error!("Fatal: failed to get ingestor info: {:?}", err);
         RBACError::Anyhow(err)
     })?;
@@ -222,7 +218,7 @@ pub async fn sync_user_creation_with_ingestors(
     user: User,
     role: &Option<HashSet<String>>,
 ) -> Result<(), RBACError> {
-    let ingestor_infos = get_ingestor_info().await.map_err(|err| {
+    let ingestor_infos: Vec<NodeMetadata> = get_node_info("ingestor").await.map_err(|err| {
         error!("Fatal: failed to get ingestor info: {:?}", err);
         RBACError::Anyhow(err)
     })?;
@@ -280,7 +276,7 @@ pub async fn sync_user_creation_with_ingestors(
 
 // forward the password reset request to all ingestors to keep them in sync
 pub async fn sync_password_reset_with_ingestors(username: &String) -> Result<(), RBACError> {
-    let ingestor_infos = get_ingestor_info().await.map_err(|err| {
+    let ingestor_infos: Vec<NodeMetadata> = get_node_info("ingestor").await.map_err(|err| {
         error!("Fatal: failed to get ingestor info: {:?}", err);
         RBACError::Anyhow(err)
     })?;
@@ -328,7 +324,7 @@ pub async fn sync_role_update_with_ingestors(
     name: String,
     privileges: Vec<DefaultPrivilege>,
 ) -> Result<(), RoleError> {
-    let ingestor_infos = get_ingestor_info().await.map_err(|err| {
+    let ingestor_infos: Vec<NodeMetadata> = get_node_info("ingestor").await.map_err(|err| {
         error!("Fatal: failed to get ingestor info: {:?}", err);
         RoleError::Anyhow(err)
     })?;
@@ -545,34 +541,48 @@ pub async fn send_retention_cleanup_request(
 
 pub async fn get_cluster_info() -> Result<impl Responder, StreamError> {
     // Get ingestor and indexer metadata concurrently
-    let (ingestor_result, indexer_result) =
-        future::join(get_ingestor_info(), get_indexer_info()).await;
+    let (querier_result, ingestor_result, indexer_result) = future::join3(
+        get_node_info("querier"),
+        get_node_info("ingestor"),
+        get_node_info("indexer"),
+    )
+    .await;
+
+    // Handle querier metadata result
+    let querier_metadata: Vec<NodeMetadata> = querier_result
+        .map_err(|err| {
+            error!("Fatal: failed to get querier info: {:?}", err);
+            PostError::Invalid(err)
+        })
+        .map_err(|err| StreamError::Anyhow(err.into()))?;
 
     // Handle ingestor metadata result
-    let ingestor_metadata = ingestor_result
+    let ingestor_metadata: Vec<NodeMetadata> = ingestor_result
         .map_err(|err| {
             error!("Fatal: failed to get ingestor info: {:?}", err);
             PostError::Invalid(err)
         })
         .map_err(|err| StreamError::Anyhow(err.into()))?;
 
     // Handle indexer metadata result
-    let indexer_metadata = indexer_result
+    let indexer_metadata: Vec<NodeMetadata> = indexer_result
         .map_err(|err| {
             error!("Fatal: failed to get indexer info: {:?}", err);
             PostError::Invalid(err)
         })
         .map_err(|err| StreamError::Anyhow(err.into()))?;
 
     // Fetch info for both node types concurrently
-    let (ingestor_infos, indexer_infos) = future::join(
+    let (querier_infos, ingestor_infos, indexer_infos) = future::join3(
+        fetch_nodes_info(querier_metadata),
         fetch_nodes_info(ingestor_metadata),
         fetch_nodes_info(indexer_metadata),
     )
     .await;
 
     // Combine results from both node types
     let mut infos = Vec::new();
+    infos.extend(querier_infos?);
     infos.extend(ingestor_infos?);
     infos.extend(indexer_infos?);
 
@@ -671,40 +681,22 @@ pub async fn get_cluster_metrics() -> Result<impl Responder, PostError> {
     Ok(actix_web::HttpResponse::Ok().json(dresses))
 }
 
-pub async fn get_ingestor_info() -> anyhow::Result<IngestorMetadataArr> {
+pub async fn get_node_info<T: Metadata + DeserializeOwned>(prefix: &str) -> anyhow::Result<Vec<T>> {
     let store = PARSEABLE.storage.get_object_store();
-
     let root_path = RelativePathBuf::from(PARSEABLE_ROOT_DIRECTORY);
-    let arr = store
-        .get_objects(
-            Some(&root_path),
-            Box::new(|file_name| file_name.starts_with("ingestor")),
-        )
-        .await?
-        .iter()
-        // this unwrap will most definateley shoot me in the foot later
-        .map(|x| serde_json::from_slice::<IngestorMetadata>(x).unwrap_or_default())
-        .collect_vec();
-
-    Ok(arr)
-}
+    let prefix_owned = prefix.to_string(); // Create an owned copy of the prefix
 
-pub async fn get_indexer_info() -> anyhow::Result<IndexerMetadataArr> {
-    let store = PARSEABLE.storage.get_object_store();
-
-    let root_path = RelativePathBuf::from(PARSEABLE_ROOT_DIRECTORY);
-    let arr = store
+    let metadata = store
         .get_objects(
             Some(&root_path),
-            Box::new(|file_name| file_name.starts_with("indexer")),
+            Box::new(move |file_name| file_name.starts_with(&prefix_owned)), // Use the owned copy
         )
         .await?
         .iter()
-        // this unwrap will most definateley shoot me in the foot later
-        .map(|x| serde_json::from_slice::<IndexerMetadata>(x).unwrap_or_default())
-        .collect_vec();
+        .filter_map(|x| serde_json::from_slice::<T>(x).ok())
+        .collect();
 
-    Ok(arr)
+    Ok(metadata)
 }
 
 pub async fn remove_node(node_url: Path<String>) -> Result<impl Responder, PostError> {
@@ -725,7 +717,11 @@ pub async fn remove_node(node_url: Path<String>) -> Result<impl Responder, PostE
     let removed_indexer =
         remove_node_metadata::<IndexerMetadata>(&object_store, &domain_name).await?;
 
-    let msg = if removed_ingestor || removed_indexer {
+    // Delete querier metadata
+    let removed_querier =
+        remove_node_metadata::<QuerierMetadata>(&object_store, &domain_name).await?;
+
+    let msg = if removed_ingestor || removed_indexer || removed_querier {
         format!("node {} removed successfully", domain_name)
     } else {
         format!("node {} is not found", domain_name)
@@ -836,6 +832,9 @@ where
     T: Metadata + Send + Sync + 'static,
 {
     let nodes_len = nodes.len();
+    if nodes_len == 0 {
+        return Ok(vec![]);
+    }
     let results = stream::iter(nodes)
         .map(|node| async move { fetch_node_metrics(&node).await })
         .buffer_unordered(nodes_len) // No concurrency limit
@@ -858,23 +857,31 @@ where
 /// Main function to fetch all cluster metrics, parallelized and refactored
 async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
     // Get ingestor and indexer metadata concurrently
-    let (ingestor_result, indexer_result) =
-        future::join(get_ingestor_info(), get_indexer_info()).await;
+    let (querier_result, ingestor_result, indexer_result) = future::join3(
+        get_node_info("querier"),
+        get_node_info("ingestor"),
+        get_node_info("indexer"),
+    )
+    .await;
 
+    // Handle querier metadata result
+    let querier_metadata: Vec<NodeMetadata> = querier_result.map_err(|err| {
+        error!("Fatal: failed to get querier info: {:?}", err);
+        PostError::Invalid(err)
+    })?;
     // Handle ingestor metadata result
-    let ingestor_metadata = ingestor_result.map_err(|err| {
+    let ingestor_metadata: Vec<NodeMetadata> = ingestor_result.map_err(|err| {
         error!("Fatal: failed to get ingestor info: {:?}", err);
         PostError::Invalid(err)
     })?;
-
     // Handle indexer metadata result
-    let indexer_metadata = indexer_result.map_err(|err| {
+    let indexer_metadata: Vec<NodeMetadata> = indexer_result.map_err(|err| {
         error!("Fatal: failed to get indexer info: {:?}", err);
         PostError::Invalid(err)
     })?;
-
     // Fetch metrics from ingestors and indexers concurrently
-    let (ingestor_metrics, indexer_metrics) = future::join(
+    let (querier_metrics, ingestor_metrics, indexer_metrics) = future::join3(
+        fetch_nodes_metrics(querier_metadata),
         fetch_nodes_metrics(ingestor_metadata),
         fetch_nodes_metrics(indexer_metadata),
     )
@@ -883,6 +890,12 @@ async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
     // Combine all metrics
     let mut all_metrics = Vec::new();
 
+    // Add querier metrics
+    match querier_metrics {
+        Ok(metrics) => all_metrics.extend(metrics),
+        Err(err) => return Err(err),
+    }
+
     // Add ingestor metrics
     match ingestor_metrics {
         Ok(metrics) => all_metrics.extend(metrics),

src/handlers/http/cluster/utils.rs

@@ -16,7 +16,10 @@
  *
  */
 
-use crate::{handlers::http::base_path_without_preceding_slash, HTTP_CLIENT};
+use crate::{
+    handlers::http::{base_path_without_preceding_slash, modal::NodeType},
+    HTTP_CLIENT,
+};
 use actix_web::http::header;
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
@@ -55,7 +58,7 @@ pub struct ClusterInfo {
     storage_path: String,
     error: Option<String>,  // error message if the ingestor is not reachable
     status: Option<String>, // status message if the ingestor is reachable
-    node_type: String,
+    node_type: NodeType,
 }
 
 impl ClusterInfo {
@@ -66,7 +69,7 @@ impl ClusterInfo {
         storage_path: String,
         error: Option<String>,
         status: Option<String>,
-        node_type: &str,
+        node_type: &NodeType,
     ) -> Self {
         Self {
             domain_name: domain_name.to_string(),
@@ -75,7 +78,7 @@ impl ClusterInfo {
             storage_path,
             error,
             status,
-            node_type: node_type.to_string(),
+            node_type: node_type.clone(),
         }
     }
 }