added hotpath as a feature flag

Author: parmesant

Cargo.toml

@@ -91,7 +91,7 @@ tokio-stream = { version = "0.1.17", features = ["fs"] }
 tokio-util = { version = "0.7" }
 
 # perf
-hotpath = { version = "0.16.0", features = [
+hotpath = { version = "0.16.0", optional = true, features = [
     "hotpath",
     "hotpath-cpu",
     "hotpath-alloc",
@@ -219,6 +219,7 @@ assets-sha1 = "a7523ef819d38678275ae165c443564b2f9a3fc1"
 
 [features]
 debug = []
+hotpath = ["dep:hotpath"]
 kafka = [
     "rdkafka",
     "rdkafka/ssl-vendored",

src/event/format/json.rs

@@ -60,7 +60,7 @@ impl EventFormat for Event {
 
     // convert the incoming json to a vector of json values
     // also extract the arrow schema, tags and metadata from the incoming json
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     fn to_data(
         self,
         schema: &HashMap<String, Arc<Field>>,

src/event/format/mod.rs

@@ -162,7 +162,7 @@ pub trait EventFormat: Sized {
     /// Returns the UTC time at ingestion
     fn get_p_timestamp(&self) -> DateTime<Utc>;
 
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     fn into_recordbatch(
         self,
         storage_schema: &HashMap<String, Arc<Field>>,

src/event/mod.rs

@@ -72,7 +72,7 @@ impl Event {
             is_first_event = self.is_first_event
         )
     )]
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     pub fn process(self) -> Result<(), EventError> {
         let mut key = get_schema_key(&self.rb.schema().fields);
         if self.time_partition.is_some() {

src/handlers/http/modal/utils/ingest_utils.rs

@@ -156,7 +156,7 @@ pub async fn flatten_and_push_logs(
     skip(json, log_source, p_custom_fields, time_partition, telemetry_type, tenant_id),
     fields(stream_name, record_count = tracing::field::Empty)
 )]
-#[hotpath::measure]
+#[cfg_attr(feature = "hotpath", hotpath::measure)]
 pub fn push_logs(
     stream_name: &str,
     json: Value,

src/lib.rs

@@ -62,12 +62,14 @@ pub use datafusion;
 pub use handlers::http::modal::{
     ParseableServer, ingest_server::IngestServer, query_server::QueryServer, server::Server,
 };
+#[cfg(feature = "hotpath")]
+pub use hotpath;
 use once_cell::sync::Lazy;
 pub use openid;
 use parseable::PARSEABLE;
 use reqwest::{Client, ClientBuilder};
-pub use {hotpath, tracing_actix_web, tracing_opentelemetry, tracing_subscriber};
 pub use {opentelemetry, opentelemetry_otlp, opentelemetry_proto, opentelemetry_sdk};
+pub use {tracing_actix_web, tracing_opentelemetry, tracing_subscriber};
 
 // It is very unlikely that panic will occur when dealing with locks.
 pub const LOCK_EXPECT: &str = "Thread shouldn't panic while holding a lock";

src/otel/metrics.rs

@@ -608,7 +608,7 @@ fn process_resource_metrics<T, S, M>(
 
 /// this function performs the custom flattening of the otel metrics
 /// and returns a `Vec` of `Value::Object` of the flattened json
-#[hotpath::measure]
+#[cfg_attr(feature = "hotpath", hotpath::measure)]
 pub fn flatten_otel_metrics(message: MetricsData, tenant_id: &str) -> Vec<Value> {
     process_resource_metrics(
         &message.resource_metrics,

src/parseable/staging/writer.rs

@@ -51,7 +51,7 @@ pub struct Writer {
 }
 
 impl Writer {
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     pub fn push_disk(
         &mut self,
         filename: String,
@@ -204,7 +204,7 @@ impl DiskWriter {
     }
 
     /// Write a single recordbatch into file
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     pub fn write(&mut self, rb: &RecordBatch) -> Result<(), StagingError> {
         self.inner.write(rb).map_err(StagingError::Arrow)
     }

src/parseable/streams.rs

@@ -139,7 +139,7 @@ impl Stream {
     }
 
     // Concatenates record batches and puts them in memory store for each event.
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     pub fn push(
         &self,
         schema_key: &str,
@@ -409,7 +409,7 @@ impl Stream {
         base.join(format!("{INPROCESS_DIR_PREFIX}{minute}"))
     }
 
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     pub fn parquet_files(&self) -> Vec<PathBuf> {
         let Ok(dir) = self.data_path.read_dir() else {
             return vec![];
@@ -470,7 +470,7 @@ impl Stream {
         skip(self, tenant_id),
         fields(stream_name = %self.stream_name)
     )]
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     pub fn prepare_parquet(
         &self,
         init_signal: bool,
@@ -550,7 +550,7 @@ impl Stream {
         Ok(())
     }
 
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     pub fn flush(&self, forced: bool) -> Result<(), StagingError> {
         let _span = info_span!("flush", stream_name = %self.stream_name, forced).entered();
         // Swap out stale writers under the lock, drop them after releasing it.
@@ -652,7 +652,7 @@ impl Stream {
     /// Bails out without sorting when either source column is missing
     /// (non-metric stream, schema drift) so the caller can write the
     /// batch unchanged.
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     fn sort_batch_for_metric_pruning(
         batch: &RecordBatch,
         time_partition_field: &str,
@@ -694,7 +694,7 @@ impl Stream {
         Ok(RecordBatch::try_new(schema, columns)?)
     }
 
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     fn prepare_metric_row_group(
         schema: Arc<Schema>,
         buffer: Vec<RecordBatch>,
@@ -772,7 +772,7 @@ impl Stream {
     /// This function reads arrow files, groups their schemas
     ///
     /// converts them into parquet files and returns a merged schema
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     pub fn convert_disk_files_to_parquet(
         &self,
         time_partition: Option<&String>,
@@ -836,7 +836,7 @@ impl Stream {
         Ok(Some(Schema::try_merge(schemas)?))
     }
 
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     fn write_parquet_part_file(
         &self,
         part_path: &Path,
@@ -941,7 +941,7 @@ impl Stream {
     }
 
     /// function to validate parquet files
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     fn is_valid_parquet_file(path: &Path, stream_name: &str) -> bool {
         // First check file size as a quick validation
         match path.metadata() {
@@ -984,7 +984,7 @@ impl Stream {
         }
     }
 
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     fn cleanup_arrow_files_and_dir(&self, arrow_files: &[PathBuf], tenant_id: &Option<String>) {
         let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         for (i, file) in arrow_files.iter().enumerate() {
@@ -1368,7 +1368,7 @@ impl Stream {
         skip(self, tenant_id),
         fields(stream_name = %self.stream_name)
     )]
-    #[hotpath::measure]
+    #[cfg_attr(feature = "hotpath", hotpath::measure)]
     pub fn flush_and_convert(
         &self,
         init_signal: bool,

src/utils/json/flatten.rs

@@ -63,7 +63,7 @@ pub enum JsonFlattenError {
 
 // Recursively flattens JSON objects and arrays, e.g. with the separator `.`, starting from the TOP
 // `{"key": "value", "nested_key": {"key":"value"}}` becomes `{"key": "value", "nested_key.key": "value"}`
-#[hotpath::measure]
+#[cfg_attr(feature = "hotpath", hotpath::measure)]
 pub fn flatten(
     nested_value: &mut Value,
     separator: &str,