add comments, remove hard coded values for node types

Author: nikhilsinhaparseable

src/analytics.rs

@@ -36,7 +36,7 @@ use crate::{
         http::{
             base_path_without_preceding_slash,
             cluster::{self, utils::check_liveness},
-            modal::NodeMetadata,
+            modal::{NodeMetadata, NodeType},
         },
         STREAM_NAME_HEADER_KEY,
     },
@@ -116,7 +116,9 @@ impl Report {
         let mut active_queries = 0;
         let mut inactive_queries = 0;
 
-        let indexer_infos: Vec<NodeMetadata> = cluster::get_node_info("indexer").await?;
+        // check liveness of indexers
+        // get the count of active and inactive indexers
+        let indexer_infos: Vec<NodeMetadata> = cluster::get_node_info(NodeType::Indexer).await?;
         for indexer in indexer_infos {
             if check_liveness(&indexer.domain_name).await {
                 active_indexers += 1;
@@ -125,7 +127,9 @@ impl Report {
             }
         }
 
-        let query_infos: Vec<NodeMetadata> = cluster::get_node_info("query").await?;
+        // check liveness of queriers
+        // get the count of active and inactive queriers
+        let query_infos: Vec<NodeMetadata> = cluster::get_node_info(NodeType::Querier).await?;
         for query in query_infos {
             if check_liveness(&query.domain_name).await {
                 active_queries += 1;
@@ -254,11 +258,14 @@ async fn fetch_ingestors_metrics(
     let mut vec = vec![];
     let mut active_ingestors = 0u64;
     let mut offline_ingestors = 0u64;
+
+    // for OSS, Query mode fetches the analytics report
+    // for Enterprise, Prism mode fetches the analytics report
     if PARSEABLE.options.mode == Mode::Query || PARSEABLE.options.mode == Mode::Prism {
         // 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("ingestor").await?;
+        let ingestor_infos: Vec<NodeMetadata> = cluster::get_node_info(NodeType::Ingestor).await?;
 
         for im in ingestor_infos {
             if !check_liveness(&im.domain_name).await {

src/catalog/mod.rs

@@ -30,7 +30,10 @@ use crate::{
     event::DEFAULT_TIMESTAMP_KEY,
     handlers::{
         self,
-        http::{base_path_without_preceding_slash, modal::NodeMetadata},
+        http::{
+            base_path_without_preceding_slash,
+            modal::{NodeMetadata, NodeType},
+        },
     },
     metrics::{EVENTS_INGESTED_DATE, EVENTS_INGESTED_SIZE_DATE, EVENTS_STORAGE_SIZE_DATE},
     option::Mode,
@@ -338,6 +341,10 @@ pub async fn remove_manifest_from_snapshot(
         meta.first_event_at = None;
         storage.put_snapshot(stream_name, meta.snapshot).await?;
     }
+
+    // retention is initiated from the querier
+    // request is forwarded to all ingestors to clean up their manifests
+    // no action required for the Index or Prism nodes
     match PARSEABLE.options.mode {
         Mode::All | Mode::Ingest => {
             Ok(get_first_event(storage.clone(), stream_name, Vec::new()).await?)
@@ -359,7 +366,6 @@ pub async fn get_first_event(
 ) -> Result<Option<String>, ObjectStorageError> {
     let mut first_event_at: String = String::default();
     match PARSEABLE.options.mode {
-        Mode::Index | Mode::Prism => unimplemented!(),
         Mode::All | Mode::Ingest => {
             // get current snapshot
             let stream_first_event = PARSEABLE.get_stream(stream_name)?.get_first_event();
@@ -410,7 +416,7 @@ pub async fn get_first_event(
         }
         Mode::Query => {
             let ingestor_metadata: Vec<NodeMetadata> =
-                handlers::http::cluster::get_node_info("ingestor")
+                handlers::http::cluster::get_node_info(NodeType::Ingestor)
                     .await
                     .map_err(|err| {
                         error!("Fatal: failed to get ingestor info: {:?}", err);
@@ -440,6 +446,7 @@ pub async fn get_first_event(
             }
             first_event_at = ingestors_first_event_at.iter().min().unwrap().to_string();
         }
+        _ => {}
     }
 
     Ok(Some(first_event_at))

src/cli.rs

@@ -447,6 +447,8 @@ impl Options {
         }
     }
 
+    /// get the address of the server
+    /// based on the mode
     pub fn get_url(&self, mode: Mode) -> Url {
         let endpoint = match mode {
             Mode::Ingest => self.get_endpoint(&self.ingestor_endpoint, "P_INGESTOR_ENDPOINT"),
@@ -458,6 +460,8 @@ impl Options {
         self.parse_endpoint(&endpoint)
     }
 
+    /// get the endpoint for the server
+    /// if env var is empty, use the address, else use the env var
     fn get_endpoint(&self, endpoint: &str, env_var: &str) -> String {
         if endpoint.is_empty() {
             self.address.to_string()
@@ -472,6 +476,9 @@ impl Options {
         }
     }
 
+    /// parse the endpoint to get the address and port
+    /// if the address is an env var, resolve it
+    /// if the port is an env var, resolve it
     fn parse_endpoint(&self, endpoint: &str) -> Url {
         let addr_parts: Vec<&str> = endpoint.split(':').collect();
 
@@ -488,6 +495,9 @@ impl Options {
         self.build_url(&format!("{}:{}", hostname, port))
     }
 
+    /// resolve the env var
+    /// if the env var is not set, panic
+    /// if the env var is set, return the value
     fn resolve_env_var(&self, value: &str) -> String {
         if let Some(env_var) = value.strip_prefix('$') {
             let resolved_value = env::var(env_var).unwrap_or_else(|_| {
@@ -510,6 +520,7 @@ impl Options {
         }
     }
 
+    /// build the url from the address
     fn build_url(&self, address: &str) -> Url {
         format!("{}://{}", self.get_scheme(), address)
             .parse::<Url>()

src/handlers/airplane.rs

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

src/handlers/http/about.rs

@@ -21,7 +21,6 @@ use serde_json::{json, Value};
 
 use crate::{
     about::{self, get_latest_release},
-    option::Mode,
     parseable::PARSEABLE,
     storage::StorageMetadata,
 };
@@ -63,12 +62,7 @@ pub async fn about() -> Json<Value> {
     let commit = current_release.commit_hash;
     let deployment_id = meta.deployment_id.to_string();
     let mode = PARSEABLE.get_server_mode_string();
-    let staging = if PARSEABLE.options.mode == Mode::Query || PARSEABLE.options.mode == Mode::Prism
-    {
-        "".to_string()
-    } else {
-        PARSEABLE.options.staging_dir().display().to_string()
-    };
+    let staging = PARSEABLE.options.staging_dir().display().to_string();
     let grpc_port = PARSEABLE.options.grpc_port;
 
     let store_endpoint = PARSEABLE.storage.get_endpoint();