add node type to cluster info and metrics

nikhilsinhaparseable · nikhilsinhaparseable · commit e42b4f798f53 · 2025-03-30T01:24:03.000-04:00
diff --git a/src/handlers/http/cluster/mod.rs b/src/handlers/http/cluster/mod.rs
@@ -564,8 +564,8 @@ pub async fn get_cluster_info() -> Result<impl Responder, StreamError> {
 
     // Fetch info for both node types concurrently
     let (ingestor_infos, indexer_infos) = future::join(
-        fetch_servers_info(ingestor_metadata),
-        fetch_servers_info(indexer_metadata),
+        fetch_nodes_info(ingestor_metadata),
+        fetch_nodes_info(indexer_metadata),
     )
     .await;
 
@@ -577,18 +577,18 @@ pub async fn get_cluster_info() -> Result<impl Responder, StreamError> {
     Ok(actix_web::HttpResponse::Ok().json(infos))
 }
 
-/// Fetches info for a single server (ingestor or indexer)
-async fn fetch_server_info<T: Metadata>(server: &T) -> Result<utils::ClusterInfo, StreamError> {
+/// Fetches info for a single node (ingestor or indexer)
+async fn fetch_node_info<T: Metadata>(node: &T) -> Result<utils::ClusterInfo, StreamError> {
     let uri = Url::parse(&format!(
         "{}{}/about",
-        server.domain_name(),
+        node.domain_name(),
         base_path_without_preceding_slash()
     ))
     .expect("should always be a valid url");
 
     let resp = HTTP_CLIENT
         .get(uri)
-        .header(header::AUTHORIZATION, server.token().to_owned())
+        .header(header::AUTHORIZATION, node.token().to_owned())
         .header(header::CONTENT_TYPE, "application/json")
         .send()
         .await;
@@ -597,13 +597,13 @@ async fn fetch_server_info<T: Metadata>(server: &T) -> Result<utils::ClusterInfo
         let status = Some(resp.status().to_string());
 
         let resp_data = resp.bytes().await.map_err(|err| {
-            error!("Fatal: failed to parse server info to bytes: {:?}", err);
+            error!("Fatal: failed to parse node info to bytes: {:?}", err);
             StreamError::Network(err)
         })?;
 
         let sp = serde_json::from_slice::<JsonValue>(&resp_data)
             .map_err(|err| {
-                error!("Fatal: failed to parse server info: {:?}", err);
+                error!("Fatal: failed to parse node info: {:?}", err);
                 StreamError::SerdeError(err)
             })?
             .get("staging")
@@ -627,23 +627,24 @@ async fn fetch_server_info<T: Metadata>(server: &T) -> Result<utils::ClusterInfo
     };
 
     Ok(utils::ClusterInfo::new(
-        server.domain_name(),
+        node.domain_name(),
         reachable,
         staging_path,
         PARSEABLE.storage.get_endpoint(),
         error,
         status,
+        node.node_type(),
     ))
 }
 
-/// Fetches info for multiple servers in parallel
-async fn fetch_servers_info<T: Metadata>(
-    servers: Vec<T>,
+/// Fetches info for multiple nodes in parallel
+async fn fetch_nodes_info<T: Metadata>(
+    nodes: Vec<T>,
 ) -> Result<Vec<utils::ClusterInfo>, StreamError> {
-    let servers_len = servers.len();
-    let results = stream::iter(servers)
-        .map(|server| async move { fetch_server_info(&server).await })
-        .buffer_unordered(servers_len) // No concurrency limit
+    let nodes_len = nodes.len();
+    let results = stream::iter(nodes)
+        .map(|node| async move { fetch_node_info(&node).await })
+        .buffer_unordered(nodes_len) // No concurrency limit
         .collect::<Vec<_>>()
         .await;
 
@@ -749,31 +750,29 @@ pub async fn remove_ingestor(ingestor: Path<String>) -> Result<impl Responder, P
     Ok((msg, StatusCode::OK))
 }
 
-/// Fetches metrics from a server (ingestor or indexer)
-async fn fetch_server_metrics<T>(server: &T) -> Result<Option<Metrics>, PostError>
+/// Fetches metrics from a node (ingestor or indexer)
+async fn fetch_node_metrics<T>(node: &T) -> Result<Option<Metrics>, PostError>
 where
     T: Metadata + Send + Sync + 'static,
 {
     // Format the metrics URL
     let uri = Url::parse(&format!(
         "{}{}/metrics",
-        server.domain_name(),
+        node.domain_name(),
         base_path_without_preceding_slash()
     ))
-    .map_err(|err| {
-        PostError::Invalid(anyhow::anyhow!("Invalid URL in server metadata: {}", err))
-    })?;
+    .map_err(|err| PostError::Invalid(anyhow::anyhow!("Invalid URL in node metadata: {}", err)))?;
 
-    // Check if the server is live
-    if !check_liveness(server.domain_name()).await {
-        warn!("Server {} is not live", server.domain_name());
+    // Check if the node is live
+    if !check_liveness(node.domain_name()).await {
+        warn!("node {} is not live", node.domain_name());
         return Ok(None);
     }
 
     // Fetch metrics
     let res = HTTP_CLIENT
         .get(uri)
-        .header(header::AUTHORIZATION, server.token())
+        .header(header::AUTHORIZATION, node.token())
         .header(header::CONTENT_TYPE, "application/json")
         .send()
         .await;
@@ -788,43 +787,43 @@ where
                 .map_err(|err| PostError::CustomError(err.to_string()))?
                 .samples;
 
-            let metrics = Metrics::from_prometheus_samples(sample, server)
+            let metrics = Metrics::from_prometheus_samples(sample, node)
                 .await
                 .map_err(|err| {
-                    error!("Fatal: failed to get server metrics: {:?}", err);
+                    error!("Fatal: failed to get node metrics: {:?}", err);
                     PostError::Invalid(err.into())
                 })?;
 
             Ok(Some(metrics))
         }
         Err(_) => {
             warn!(
-                "Failed to fetch metrics from server: {}\n",
-                server.domain_name()
+                "Failed to fetch metrics from node: {}\n",
+                node.domain_name()
             );
             Ok(None)
         }
     }
 }
 
-/// Fetches metrics from multiple servers in parallel
-async fn fetch_servers_metrics<T>(servers: Vec<T>) -> Result<Vec<Metrics>, PostError>
+/// Fetches metrics from multiple nodes in parallel
+async fn fetch_nodes_metrics<T>(nodes: Vec<T>) -> Result<Vec<Metrics>, PostError>
 where
     T: Metadata + Send + Sync + 'static,
 {
-    let servers_len = servers.len();
-    let results = stream::iter(servers)
-        .map(|server| async move { fetch_server_metrics(&server).await })
-        .buffer_unordered(servers_len) // No concurrency limit
+    let nodes_len = nodes.len();
+    let results = stream::iter(nodes)
+        .map(|node| async move { fetch_node_metrics(&node).await })
+        .buffer_unordered(nodes_len) // No concurrency limit
         .collect::<Vec<_>>()
         .await;
 
     // Process results
     let mut metrics = Vec::new();
     for result in results {
         match result {
-            Ok(Some(server_metrics)) => metrics.push(server_metrics),
-            Ok(None) => {} // server was not live or metrics couldn't be fetched
+            Ok(Some(node_metrics)) => metrics.push(node_metrics),
+            Ok(None) => {} // node was not live or metrics couldn't be fetched
             Err(err) => return Err(err),
         }
     }
@@ -852,8 +851,8 @@ async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
 
     // Fetch metrics from ingestors and indexers concurrently
     let (ingestor_metrics, indexer_metrics) = future::join(
-        fetch_servers_metrics(ingestor_metadata),
-        fetch_servers_metrics(indexer_metadata),
+        fetch_nodes_metrics(ingestor_metadata),
+        fetch_nodes_metrics(indexer_metadata),
     )
     .await;
 
diff --git a/src/handlers/http/cluster/utils.rs b/src/handlers/http/cluster/utils.rs
@@ -55,6 +55,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,
 }
 
 impl ClusterInfo {
@@ -65,6 +66,7 @@ impl ClusterInfo {
         storage_path: String,
         error: Option<String>,
         status: Option<String>,
+        node_type: &str,
     ) -> Self {
         Self {
             domain_name: domain_name.to_string(),
@@ -73,6 +75,7 @@ impl ClusterInfo {
             storage_path,
             error,
             status,
+            node_type: node_type.to_string(),
         }
     }
 }
diff --git a/src/handlers/http/modal/mod.rs b/src/handlers/http/modal/mod.rs
@@ -552,6 +552,7 @@ impl IndexerMetadata {
 pub trait Metadata {
     fn domain_name(&self) -> &str;
     fn token(&self) -> &str;
+    fn node_type(&self) -> &str;
 }
 
 impl Metadata for IngestorMetadata {
@@ -562,6 +563,9 @@ impl Metadata for IngestorMetadata {
     fn token(&self) -> &str {
         &self.token
     }
+    fn node_type(&self) -> &str {
+        "ingestor"
+    }
 }
 
 impl Metadata for IndexerMetadata {
@@ -572,6 +576,9 @@ impl Metadata for IndexerMetadata {
     fn token(&self) -> &str {
         &self.token
     }
+    fn node_type(&self) -> &str {
+        "indexer"
+    }
 }
 #[cfg(test)]
 mod test {
diff --git a/src/metrics/prom_utils.rs b/src/metrics/prom_utils.rs
@@ -37,6 +37,7 @@ use url::Url;
 #[derive(Debug, Serialize, Clone)]
 pub struct Metrics {
     address: String,
+    node_type: String,
     parseable_events_ingested: f64, // all streams
     parseable_events_ingested_size: f64,
     parseable_lifetime_events_ingested: f64, // all streams
@@ -72,6 +73,7 @@ impl Default for Metrics {
         );
         Metrics {
             address,
+            node_type: "ingestor".to_string(),
             parseable_events_ingested: 0.0,
             parseable_events_ingested_size: 0.0,
             parseable_staging_files: 0.0,
@@ -92,9 +94,10 @@ impl Default for Metrics {
 }
 
 impl Metrics {
-    fn new(address: String) -> Self {
+    fn new(address: String, node_type: String) -> Self {
         Metrics {
             address,
+            node_type,
             parseable_events_ingested: 0.0,
             parseable_events_ingested_size: 0.0,
             parseable_staging_files: 0.0,
@@ -160,7 +163,10 @@ impl Metrics {
         samples: Vec<PromSample>,
         metadata: &T,
     ) -> Result<Self, PostError> {
-        let mut prom_dress = Metrics::new(metadata.domain_name().to_string());
+        let mut prom_dress = Metrics::new(
+            metadata.domain_name().to_string(),
+            metadata.node_type().to_string(),
+        );
         for sample in samples {
             if let PromValue::Gauge(val) = sample.value {
                 match sample.metric.as_str() {

Original file line number	Diff line number	Diff line change
`@@ -55,6 +55,7 @@ pub struct ClusterInfo {`
`55`	`55`	`storage_path: String,`
`56`	`56`	`error: Option<String>, // error message if the ingestor is not reachable`
`57`	`57`	`status: Option<String>, // status message if the ingestor is reachable`
	`58`	`+ node_type: String,`
`58`	`59`	`}`
`59`	`60`
`60`	`61`	`impl ClusterInfo {`
`@@ -65,6 +66,7 @@ impl ClusterInfo {`
`65`	`66`	`storage_path: String,`
`66`	`67`	`error: Option<String>,`
`67`	`68`	`status: Option<String>,`
	`69`	`+ node_type: &str,`
`68`	`70`	`) -> Self {`
`69`	`71`	`Self {`
`70`	`72`	`domain_name: domain_name.to_string(),`
`@@ -73,6 +75,7 @@ impl ClusterInfo {`
`73`	`75`	`storage_path,`
`74`	`76`	`error,`
`75`	`77`	`status,`
	`78`	`+ node_type: node_type.to_string(),`
`76`	`79`	`}`
`77`	`80`	`}`
`78`	`81`	`}`
Original file line number	Diff line number	Diff line change
`@@ -552,6 +552,7 @@ impl IndexerMetadata {`
`552`	`552`	`pub trait Metadata {`
`553`	`553`	`fn domain_name(&self) -> &str;`
`554`	`554`	`fn token(&self) -> &str;`
	`555`	`+ fn node_type(&self) -> &str;`
`555`	`556`	`}`
`556`	`557`
`557`	`558`	`impl Metadata for IngestorMetadata {`
`@@ -562,6 +563,9 @@ impl Metadata for IngestorMetadata {`
`562`	`563`	`fn token(&self) -> &str {`
`563`	`564`	`&self.token`
`564`	`565`	`}`
	`566`	`+ fn node_type(&self) -> &str {`
	`567`	`+ "ingestor"`
	`568`	`+ }`
`565`	`569`	`}`
`566`	`570`
`567`	`571`	`impl Metadata for IndexerMetadata {`
`@@ -572,6 +576,9 @@ impl Metadata for IndexerMetadata {`
`572`	`576`	`fn token(&self) -> &str {`
`573`	`577`	`&self.token`
`574`	`578`	`}`
	`579`	`+ fn node_type(&self) -> &str {`
	`580`	`+ "indexer"`
	`581`	`+ }`
`575`	`582`	`}`
`576`	`583`	`#[cfg(test)]`
`577`	`584`	`mod test {`