feat: add sysinfo metrics

collect CPU usage, memory usage of the server
collect disk usage of the volume - data, staging, hot-tier
add these metrics to Prometheus Metrics

export these metrics to cluster metrics API
add the metrics to pmeta stream
add the querier node's sysinfo metrics to pmeta and cluster metrics API

Author: nikhilsinhaparseable

src/correlation.rs

@@ -129,10 +129,9 @@ impl Correlations {
             .await?;
 
         // Update in memory
-        self.write().await.insert(
-            correlation.id.to_owned(),
-            correlation.clone(),
-        );
+        self.write()
+            .await
+            .insert(correlation.id.to_owned(), correlation.clone());
 
         Ok(correlation)
     }

src/handlers/http/cluster/mod.rs

@@ -35,7 +35,7 @@ use crate::storage::{ObjectStorageError, STREAM_ROOT_DIRECTORY};
 use crate::storage::{ObjectStoreFormat, PARSEABLE_ROOT_DIRECTORY};
 use crate::HTTP_CLIENT;
 use actix_web::http::header::{self, HeaderMap};
-use actix_web::web::Path;
+use actix_web::web::{Json, Path};
 use actix_web::Responder;
 use bytes::Bytes;
 use chrono::Utc;
@@ -729,6 +729,9 @@ async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
     let mut dresses = vec![];
 
     for ingestor in ingestor_metadata {
+        if !utils::check_liveness(&ingestor.domain_name).await {
+            continue;
+        }
         let uri = Url::parse(&format!(
             "{}{}/metrics",
             &ingestor.domain_name,
@@ -749,11 +752,10 @@ async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
             let text = res.text().await.map_err(PostError::NetworkError)?;
             let lines: Vec<Result<String, std::io::Error>> =
                 text.lines().map(|line| Ok(line.to_owned())).collect_vec();
-
             let sample = prometheus_parse::Scrape::parse(lines.into_iter())
                 .map_err(|err| PostError::CustomError(err.to_string()))?
                 .samples;
-            let ingestor_metrics = Metrics::from_prometheus_samples(sample, &ingestor)
+            let ingestor_metrics = Metrics::ingestor_prometheus_samples(sample, &ingestor)
                 .await
                 .map_err(|err| {
                     error!("Fatal: failed to get ingestor metrics: {:?}", err);
@@ -767,10 +769,11 @@ async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
             );
         }
     }
+    dresses.push(Metrics::querier_prometheus_metrics().await);
     Ok(dresses)
 }
 
-pub fn init_cluster_metrics_schedular() -> Result<(), PostError> {
+pub async fn init_cluster_metrics_scheduler() -> Result<(), PostError> {
     info!("Setting up schedular for cluster metrics ingestion");
     let mut scheduler = AsyncScheduler::new();
     scheduler
@@ -779,25 +782,12 @@ pub fn init_cluster_metrics_schedular() -> Result<(), PostError> {
             let result: Result<(), PostError> = async {
                 let cluster_metrics = fetch_cluster_metrics().await;
                 if let Ok(metrics) = cluster_metrics {
-                    if !metrics.is_empty() {
-                        info!("Cluster metrics fetched successfully from all ingestors");
-                        if let Ok(metrics_bytes) = serde_json::to_vec(&metrics) {
-                            if matches!(
-                                ingest_internal_stream(
-                                    INTERNAL_STREAM_NAME.to_string(),
-                                    bytes::Bytes::from(metrics_bytes),
-                                )
-                                .await,
-                                Ok(())
-                            ) {
-                                info!("Cluster metrics successfully ingested into internal stream");
-                            } else {
-                                error!("Failed to ingest cluster metrics into internal stream");
-                            }
-                        } else {
-                            error!("Failed to serialize cluster metrics");
-                        }
-                    }
+                    let json_value = serde_json::to_value(metrics)
+                        .map_err(|e| anyhow::anyhow!("Failed to serialize metrics: {}", e))?;
+
+                    ingest_internal_stream(INTERNAL_STREAM_NAME.to_string(), Json(json_value))
+                        .await
+                        .map_err(|e| anyhow::anyhow!("Failed to ingest metrics: {}", e))?;
                 }
                 Ok(())
             }

src/handlers/http/ingest.rs

@@ -20,12 +20,8 @@ use super::logstream::error::{CreateStreamError, StreamError};
 use super::modal::utils::ingest_utils::{flatten_and_push_logs, push_logs};
 use super::users::dashboards::DashboardError;
 use super::users::filters::FiltersError;
-use crate::event::format::LogSource;
-use crate::event::{
-    self,
-    error::EventError,
-    format::{self, EventFormat},
-};
+use crate::event::format::{self, EventFormat, LogSource};
+use crate::event::{self, error::EventError};
 use crate::handlers::http::modal::utils::logstream_utils::create_stream_and_schema_from_storage;
 use crate::handlers::{LOG_SOURCE_KEY, STREAM_NAME_HEADER_KEY};
 use crate::metadata::error::stream_info::MetadataError;
@@ -36,12 +32,12 @@ use crate::otel::metrics::flatten_otel_metrics;
 use crate::otel::traces::flatten_otel_traces;
 use crate::storage::{ObjectStorageError, StreamType};
 use crate::utils::header_parsing::ParseHeaderError;
-use crate::utils::json::flatten::JsonFlattenError;
+use crate::utils::json::convert_array_to_object;
+use crate::utils::json::flatten::{convert_to_array, JsonFlattenError};
 use actix_web::web::{Json, Path};
 use actix_web::{http::header::ContentType, HttpRequest, HttpResponse};
 use arrow_array::RecordBatch;
 use arrow_schema::Schema;
-use bytes::Bytes;
 use chrono::Utc;
 use http::StatusCode;
 use opentelemetry_proto::tonic::logs::v1::LogsData;
@@ -77,18 +73,29 @@ pub async fn ingest(req: HttpRequest, Json(json): Json<Value>) -> Result<HttpRes
     Ok(HttpResponse::Ok().finish())
 }
 
-pub async fn ingest_internal_stream(stream_name: String, body: Bytes) -> Result<(), PostError> {
-    let size: usize = body.len();
+pub async fn ingest_internal_stream(
+    stream_name: String,
+    Json(json): Json<Value>,
+) -> Result<(), PostError> {
+    let size = serde_json::to_vec(&json).unwrap().len() as u64;
+    let data = convert_to_array(convert_array_to_object(
+        json,
+        None,
+        None,
+        None,
+        SchemaVersion::V0,
+        &LogSource::default(),
+    )?)?;
+
     let parsed_timestamp = Utc::now().naive_utc();
     let (rb, is_first) = {
-        let body_val: Value = serde_json::from_slice(&body)?;
         let hash_map = STREAM_INFO.read().unwrap();
         let schema = hash_map
             .get(&stream_name)
             .ok_or(PostError::StreamNotFound(stream_name.clone()))?
             .schema
             .clone();
-        let event = format::json::Event { data: body_val };
+        let event = format::json::Event { data };
         // For internal streams, use old schema
         event.into_recordbatch(&schema, false, None, SchemaVersion::V0)?
     };

src/handlers/http/modal/ingest_server.rs

@@ -113,6 +113,7 @@ impl ParseableServer for IngestServer {
 
         // set the ingestor metadata
         set_ingestor_metadata().await?;
+        metrics::init_system_metrics_scheduler().await?;
 
         // Ingestors shouldn't have to deal with OpenId auth flow
         let app = self.start(shutdown_rx, prometheus, None);

src/handlers/http/modal/query_server.rs

@@ -18,11 +18,10 @@
 
 use crate::correlation::CORRELATIONS;
 use crate::handlers::airplane;
-use crate::handlers::http::base_path;
-use crate::handlers::http::cluster::{self, init_cluster_metrics_schedular};
 use crate::handlers::http::logstream::create_internal_stream_if_not_exists;
 use crate::handlers::http::middleware::{DisAllowRootUser, RouteExt};
 use crate::handlers::http::{self, role};
+use crate::handlers::http::{base_path, cluster};
 use crate::handlers::http::{logstream, MAX_EVENT_PAYLOAD_SIZE};
 use crate::hottier::HotTierManager;
 use crate::rbac::role::Action;
@@ -35,7 +34,7 @@ use actix_web::{web, Scope};
 use async_trait::async_trait;
 use bytes::Bytes;
 use tokio::sync::oneshot;
-use tracing::{error, info};
+use tracing::error;
 
 use crate::{option::CONFIG, ParseableServer};
 
@@ -104,15 +103,14 @@ impl ParseableServer for QueryServer {
         // track all parquet files already in the data directory
         storage::retention::load_retention_from_global();
 
+        metrics::init_system_metrics_scheduler().await?;
+        cluster::init_cluster_metrics_scheduler().await?;
         // all internal data structures populated now.
         // start the analytics scheduler if enabled
         if CONFIG.options.send_analytics {
             analytics::init_analytics_scheduler()?;
         }
 
-        if matches!(init_cluster_metrics_schedular(), Ok(())) {
-            info!("Cluster metrics scheduler started successfully");
-        }
         if let Some(hot_tier_manager) = HotTierManager::global() {
             hot_tier_manager.put_internal_stream_hot_tier().await?;
             hot_tier_manager.download_from_s3()?;

src/handlers/http/modal/server.rs

@@ -119,6 +119,8 @@ impl ParseableServer for Server {
         let (mut remote_sync_handler, mut remote_sync_outbox, mut remote_sync_inbox) =
             sync::object_store_sync().await;
 
+        metrics::init_system_metrics_scheduler().await?;
+
         if CONFIG.options.send_analytics {
             analytics::init_analytics_scheduler()?;
         }