merge from main

Author: nikhilsinhaparseable

src/cli.rs

@@ -441,81 +441,77 @@ impl Options {
 
     /// TODO: refactor and document
     pub fn get_url(&self, mode: Mode) -> Url {
-        let (endpoint, env_var) = match mode {
-            Mode::Ingest => {
-                if self.ingestor_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.ingestor_endpoint, "P_INGESTOR_ENDPOINT")
-            }
-            Mode::Index => {
-                if self.indexer_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.indexer_endpoint, "P_INDEXER_ENDPOINT")
-            }
-            _ => panic!("Invalid mode"),
+        let endpoint = match mode {
+            Mode::Ingest => self.get_endpoint(&self.ingestor_endpoint, "P_INGESTOR_ENDPOINT"),
+            Mode::Index => self.get_endpoint(&self.indexer_endpoint, "P_INDEXER_ENDPOINT"),
+            Mode::Query | Mode::All => return self.build_url(&self.address),
         };
 
-        if endpoint.starts_with("http") {
-            panic!("Invalid value `{}`, please set the environement variable `{env_var}` to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details.", endpoint);
-        }
+        self.parse_endpoint(&endpoint)
+    }
 
-        let addr_from_env = endpoint.split(':').collect::<Vec<&str>>();
+    fn get_endpoint(&self, endpoint: &str, env_var: &str) -> String {
+        if endpoint.is_empty() {
+            return self.address.clone();
+        }
 
-        if addr_from_env.len() != 2 {
-            panic!("Invalid value `{}`, please set the environement variable `{env_var}` to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details.", endpoint);
+        if endpoint.starts_with("http") {
+            panic!(
+                "Invalid value `{}`, please set the environment variable `{}` to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details.",
+                endpoint, env_var
+            );
         }
 
-        let mut hostname = addr_from_env[0].to_string();
-        let mut port = addr_from_env[1].to_string();
+        endpoint.to_string()
+    }
 
-        // if the env var value fits the pattern $VAR_NAME:$VAR_NAME
-        // fetch the value from the specified env vars
-        if hostname.starts_with('$') {
-            let var_hostname = hostname[1..].to_string();
-            hostname = env::var(&var_hostname).unwrap_or_default();
+    fn parse_endpoint(&self, endpoint: &str) -> Url {
+        let addr_parts: Vec<&str> = endpoint.split(':').collect();
 
-            if hostname.is_empty() {
-                panic!("The environement variable `{}` is not set, please set as <ip address / DNS> without the scheme (e.g., 192.168.1.1 or example.com). Please refer to the documentation: https://logg.ing/env for more details.", var_hostname);
-            }
-            if hostname.starts_with("http") {
-                panic!("Invalid value `{}`, please set the environement variable `{}` to `<ip address / DNS>` without the scheme (e.g., 192.168.1.1 or example.com). Please refer to the documentation: https://logg.ing/env for more details.", hostname, var_hostname);
-            } else {
-                hostname = format!("{}://{}", self.get_scheme(), hostname);
-            }
+        if addr_parts.len() != 2 {
+            panic!(
+                "Invalid value `{}`, please set the environment variable to `<ip address / DNS>:<port>` without the scheme (e.g., 192.168.1.1:8000 or example.com:8000). Please refer to the documentation: https://logg.ing/env for more details.",
+                endpoint
+            );
         }
 
-        if port.starts_with('$') {
-            let var_port = port[1..].to_string();
-            port = env::var(&var_port).unwrap_or_default();
+        let hostname = self.resolve_env_var(addr_parts[0]);
+        let port = self.resolve_env_var(addr_parts[1]);
 
-            if port.is_empty() {
+        self.build_url(&format!("{}:{}", hostname, port))
+    }
+
+    fn resolve_env_var(&self, value: &str) -> String {
+        if let Some(stripped) = value.strip_prefix('$') {
+            let var_name = stripped;
+            let resolved_value = env::var(var_name).unwrap_or_else(|_| {
+                panic!(
+                    "The environment variable `{}` is not set. Please set it to a valid value. Refer to the documentation: https://logg.ing/env for more details.",
+                    var_name
+                );
+            });
+
+            if resolved_value.starts_with("http") {
                 panic!(
-                    "Port is not set in the environement variable `{}`. Please refer to the documentation: https://logg.ing/env for more details.",
-                    var_port
+                    "Invalid value `{}`, please set the environment variable `{}` to `<ip address / DNS>` without the scheme (e.g., 192.168.1.1 or example.com). Please refer to the documentation: https://logg.ing/env for more details.",
+                    resolved_value, var_name
                 );
             }
+
+            resolved_value
+        } else {
+            value.to_string()
         }
+    }
 
-        format!("{}://{}:{}", self.get_scheme(), hostname, port)
+    fn build_url(&self, address: &str) -> Url {
+        format!("{}://{}", self.get_scheme(), address)
             .parse::<Url>()
-            .expect("Valid URL")
+            .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.",
+                    address
+                );
+            })
     }
 }

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

@@ -23,7 +23,7 @@ use std::collections::HashSet;
 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;
@@ -39,6 +39,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;
@@ -870,6 +871,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)
 }
 
@@ -880,14 +882,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

@@ -28,6 +28,7 @@ use relative_path::RelativePathBuf;
 use serde_json::Value;
 use tokio::sync::oneshot;
 
+use crate::metrics::init_system_metrics_scheduler;
 use crate::option::Mode;
 use crate::{
     analytics,
@@ -110,7 +111,7 @@ impl ParseableServer for IngestServer {
 
         // write the ingestor metadata to storage
         PARSEABLE.store_metadata(Mode::Ingest).await?;
-
+        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

@@ -19,12 +19,13 @@
 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};
@@ -33,7 +34,6 @@ use actix_web_prometheus::PrometheusMetrics;
 use async_trait::async_trait;
 use bytes::Bytes;
 use tokio::sync::oneshot;
-use tracing::info;
 
 use crate::parseable::PARSEABLE;
 use crate::Server;
@@ -109,17 +109,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)?;
     }