merge from main

Author: nikhilsinhaparseable

src/event/mod.rs

@@ -76,7 +76,6 @@ impl Event {
             &self.rb,
             self.parsed_timestamp,
             &self.custom_partition_values,
-            self.stream_type,
         )?;
 
         update_stats(
@@ -100,7 +99,6 @@ impl Event {
             &self.rb,
             self.parsed_timestamp,
             &self.custom_partition_values,
-            self.stream_type,
         )?;
 
         Ok(())

src/handlers/http/cluster/mod.rs

@@ -25,7 +25,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use actix_web::http::header::{self, HeaderMap};
-use actix_web::web::{Json, Path};
+use actix_web::web::Path;
 use actix_web::Responder;
 use bytes::Bytes;
 use chrono::Utc;
@@ -41,6 +41,7 @@ use url::Url;
 use utils::{check_liveness, to_url_string, IngestionStats, QueriedStats, StorageStats};
 
 use crate::handlers::http::ingest::ingest_internal_stream;
+use crate::metrics::collect_all_metrics;
 use crate::metrics::prom_utils::Metrics;
 use crate::parseable::PARSEABLE;
 use crate::rbac::role::model::DefaultPrivilege;
@@ -991,6 +992,7 @@ async fn fetch_cluster_metrics() -> Result<Vec<Metrics>, PostError> {
         Err(err) => return Err(err),
     }
 
+    all_metrics.push(Metrics::querier_prometheus_metrics().await);
     Ok(all_metrics)
 }
 
@@ -1001,14 +1003,30 @@ pub async fn init_cluster_metrics_scheduler() -> Result<(), PostError> {
         .every(CLUSTER_METRICS_INTERVAL_SECONDS)
         .run(move || async {
             let result: Result<(), PostError> = async {
+                if let Err(err) = collect_all_metrics().await {
+                    error!("Error in capturing system metrics: {:#}", err);
+                }
                 let cluster_metrics = fetch_cluster_metrics().await;
                 if let Ok(metrics) = 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))?;
+                    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");
+                        }
+                    }
                 }
                 Ok(())
             }

src/handlers/http/ingest.rs

@@ -18,37 +18,19 @@
 
 use std::collections::{HashMap, HashSet};
 
-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::{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;
-use crate::metadata::{SchemaVersion, STREAM_INFO};
-use crate::option::{Mode, CONFIG};
-use crate::otel::logs::flatten_otel_logs;
-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::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 bytes::Bytes;
 use chrono::Utc;
 use http::StatusCode;
 use serde_json::Value;
 
 use crate::event::error::EventError;
 use crate::event::format::known_schema::{self, KNOWN_SCHEMA_LIST};
-use crate::event::format::{self, EventFormat, LogSource, LogSourceEntry};
+use crate::event::format::{LogSource, LogSourceEntry};
 use crate::event::{self, FORMAT_KEY, USER_AGENT_KEY};
 use crate::handlers::{EXTRACT_LOG_KEY, LOG_SOURCE_KEY, STREAM_NAME_HEADER_KEY};
-use crate::metadata::SchemaVersion;
 use crate::option::Mode;
 use crate::otel::logs::OTEL_LOG_KNOWN_FIELD_LIST;
 use crate::otel::metrics::OTEL_METRICS_KNOWN_FIELD_LIST;
@@ -138,26 +120,11 @@ pub async fn ingest(
 }
 
 pub async fn ingest_internal_stream(stream_name: String, body: Bytes) -> Result<(), PostError> {
-    let size: usize = body.len();
     let json: Value = serde_json::from_slice(&body)?;
-    let schema = PARSEABLE.get_stream(&stream_name)?.get_schema_raw();
     let mut p_custom_fields = HashMap::new();
     p_custom_fields.insert(USER_AGENT_KEY.to_string(), "parseable".to_string());
     p_custom_fields.insert(FORMAT_KEY.to_string(), LogSource::Pmeta.to_string());
-    // For internal streams, use old schema
-    format::json::Event::new(json)
-        .into_event(
-            stream_name,
-            size as u64,
-            &schema,
-            false,
-            None,
-            None,
-            SchemaVersion::V0,
-            StreamType::Internal,
-            &p_custom_fields,
-        )?
-        .process()?;
+    flatten_and_push_logs(json, &stream_name, &LogSource::Pmeta, &p_custom_fields).await?;
 
     Ok(())
 }

src/handlers/http/modal/ingest_server.rs

@@ -30,6 +30,7 @@ use serde_json::Value;
 use tokio::sync::oneshot;
 use tokio::sync::OnceCell;
 
+use crate::metrics::init_system_metrics_scheduler;
 use crate::handlers::http::modal::NodeType;
 use crate::{
     analytics,
@@ -119,7 +120,7 @@ impl ParseableServer for IngestServer {
         thread::spawn(|| sync::handler(cancel_rx));
 
         tokio::spawn(airplane::server());
-
+        init_system_metrics_scheduler().await?;
         // Ingestors shouldn't have to deal with OpenId auth flow
         let result = self.start(shutdown_rx, prometheus.clone(), None).await;
         // Cancel sync jobs

src/handlers/http/modal/query_server.rs

@@ -20,12 +20,13 @@ use std::sync::Arc;
 use std::thread;
 
 use crate::handlers::airplane;
-use crate::handlers::http::cluster::{self, init_cluster_metrics_schedular};
+use crate::handlers::http::cluster::{self, init_cluster_metrics_scheduler};
 use crate::handlers::http::middleware::{DisAllowRootUser, RouteExt};
 use crate::handlers::http::{base_path, prism_base_path};
 use crate::handlers::http::{logstream, MAX_EVENT_PAYLOAD_SIZE};
 use crate::handlers::http::{rbac, role};
 use crate::hottier::HotTierManager;
+use crate::metrics::init_system_metrics_scheduler;
 use crate::rbac::role::Action;
 use crate::{analytics, migration, storage, sync};
 use actix_web::web::{resource, ServiceConfig};
@@ -34,7 +35,6 @@ use actix_web_prometheus::PrometheusMetrics;
 use async_trait::async_trait;
 use bytes::Bytes;
 use tokio::sync::{oneshot, OnceCell};
-use tracing::info;
 
 use crate::parseable::PARSEABLE;
 use crate::Server;
@@ -117,17 +117,15 @@ 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 PARSEABLE.options.send_analytics {
             analytics::init_analytics_scheduler()?;
         }
 
-        if init_cluster_metrics_schedular().is_ok() {
-            info!("Cluster metrics scheduler started successfully");
-        }
+        init_system_metrics_scheduler().await?;
+        init_cluster_metrics_scheduler().await?;
+
         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

@@ -30,6 +30,7 @@ use crate::handlers::http::users::dashboards;
 use crate::handlers::http::users::filters;
 use crate::hottier::HotTierManager;
 use crate::metrics;
+use crate::metrics::init_system_metrics_scheduler;
 use crate::migration;
 use crate::storage;
 use crate::sync;
@@ -134,6 +135,8 @@ impl ParseableServer for Server {
             analytics::init_analytics_scheduler()?;
         }
 
+        init_system_metrics_scheduler().await?;
+
         tokio::spawn(handlers::livetail::server());
         tokio::spawn(handlers::airplane::server());
 

src/metrics/mod.rs

@@ -26,7 +26,7 @@ use std::{path::Path, time::Duration};
 use sysinfo::{Disks, System};
 use tracing::{error, info};
 
-use crate::{handlers::http::metrics_path, option::CONFIG, stats::FullStats};
+use crate::{handlers::http::metrics_path, parseable::PARSEABLE, stats::FullStats};
 use actix_web::Responder;
 use actix_web_prometheus::{PrometheusMetrics, PrometheusMetricsBuilder};
 use error::MetricsError;
@@ -419,14 +419,14 @@ pub async fn collect_all_metrics() -> Result<(), MetricsError> {
 // Function to collect disk usage metrics
 async fn collect_disk_metrics() -> Result<(), MetricsError> {
     // collect staging volume metrics
-    collect_volume_disk_usage("staging", CONFIG.staging_dir())?;
+    collect_volume_disk_usage("staging", PARSEABLE.options.staging_dir())?;
     // Collect data volume metrics for local storage
-    if CONFIG.get_storage_mode_string() == "Local drive" {
-        collect_volume_disk_usage("data", Path::new(&CONFIG.storage().get_endpoint()))?;
+    if PARSEABLE.get_storage_mode_string() == "Local drive" {
+        collect_volume_disk_usage("data", Path::new(&PARSEABLE.storage().get_endpoint()))?;
     }
 
     // Collect hot tier volume metrics if configured
-    if let Some(hot_tier_dir) = CONFIG.hot_tier_dir() {
+    if let Some(hot_tier_dir) = PARSEABLE.hot_tier_dir() {
         collect_volume_disk_usage("hot_tier", hot_tier_dir)?;
     }