perf: improve otel metrics ingest path

  Batch staging arrow writes per output file and prepare OTEL metric
  parquet row groups off-thread before sequential writes. This reduces
  request-path writer contention and hides row-group concat/sort work
  behind merge/write.

  Reduce JSON allocation churn in OTEL metric flattening and generic
  flattening by reusing owned maps, pre-sizing containers, inserting
  attributes/exemplars directly, and avoiding per-row known-field set
  construction for series hashing.

  Also guard shutdown local sync against concurrent local sync cycles and
  avoid panicking on transient arrow-file metadata races.

Author: parmesant

src/handlers/http/health_check.rs

@@ -32,6 +32,7 @@ use once_cell::sync::Lazy;
 use tokio::{sync::Mutex, task::JoinSet};
 use tracing::{error, info};
 
+use crate::sync::shutdown_local_sync_flush_and_convert;
 use crate::utils::get_tenant_id_from_request;
 use crate::{parseable::PARSEABLE, storage::object_storage::sync_all_streams};
 
@@ -84,20 +85,7 @@ async fn perform_sync_operations() {
 }
 
 async fn perform_local_sync() {
-    let mut local_sync_joinset = JoinSet::new();
-
-    // Sync staging
-    PARSEABLE
-        .streams
-        .flush_and_convert(&mut local_sync_joinset, false, true);
-
-    while let Some(res) = local_sync_joinset.join_next().await {
-        match res {
-            Ok(Ok(_)) => info!("Successfully converted arrow files to parquet."),
-            Ok(Err(err)) => error!("Failed to convert arrow files to parquet. {err:?}"),
-            Err(err) => error!("Failed to join async task: {err}"),
-        }
-    }
+    shutdown_local_sync_flush_and_convert().await;
 }
 
 async fn perform_object_store_sync() {

src/otel/logs.rs

@@ -138,7 +138,7 @@ pub fn flatten_log_record(log_record: &LogRecord) -> Map<String, Value> {
 
     if log_record.body.is_some() {
         let body = &log_record.body;
-        let body_json = collect_json_from_values(body, &"body".to_string());
+        let body_json = collect_json_from_values(body, "body");
         for (key, value) in &body_json {
             // Always insert the original body field as is
             log_record_json.insert(key.clone(), value.clone());

src/otel/metrics.rs

@@ -23,21 +23,26 @@ use opentelemetry_proto::tonic::common::v1::{
 use serde_json::{Map, Value};
 
 // Value can be one of types - String, Bool, Int, Double, ArrayValue, AnyValue, KeyValueList, Byte
-pub fn collect_json_from_value(key: &String, value: OtelValue) -> Map<String, Value> {
+pub fn collect_json_from_value(key: &str, value: OtelValue) -> Map<String, Value> {
     let mut value_json: Map<String, Value> = Map::new();
+    insert_json_from_value(&mut value_json, key, value);
+    value_json
+}
+
+fn insert_json_from_value(map: &mut Map<String, Value>, key: &str, value: OtelValue) {
     match value {
         OtelValue::StringValue(str_val) => {
-            value_json.insert(key.to_string(), Value::String(str_val));
+            map.insert(key.to_string(), Value::String(str_val));
         }
         OtelValue::BoolValue(bool_val) => {
-            value_json.insert(key.to_string(), Value::Bool(bool_val));
+            map.insert(key.to_string(), Value::Bool(bool_val));
         }
         OtelValue::IntValue(int_val) => {
-            value_json.insert(key.to_string(), Value::String(int_val.to_string()));
+            map.insert(key.to_string(), Value::String(int_val.to_string()));
         }
         OtelValue::DoubleValue(double_val) => {
             if let Some(number) = serde_json::Number::from_f64(double_val) {
-                value_json.insert(key.to_string(), Value::Number(number));
+                map.insert(key.to_string(), Value::Number(number));
             }
         }
         OtelValue::ArrayValue(array_val) => {
@@ -51,17 +56,15 @@ pub fn collect_json_from_value(key: &String, value: OtelValue) -> Map<String, Va
                 }
             };
             // Insert the array into the result map
-            value_json.insert(key.to_string(), Value::String(json_array_string));
+            map.insert(key.to_string(), Value::String(json_array_string));
         }
         OtelValue::KvlistValue(kv_list_val) => {
             // Create a JSON object to store the key-value list
             let kv_object = collect_json_from_key_value_list(kv_list_val);
-            for (key, value) in kv_object.iter() {
-                value_json.insert(key.clone(), value.clone());
-            }
+            map.extend(kv_object);
         }
         OtelValue::BytesValue(bytes_val) => {
-            value_json.insert(
+            map.insert(
                 key.to_string(),
                 Value::String(String::from_utf8_lossy(&bytes_val).to_string()),
             );
@@ -72,15 +75,13 @@ pub fn collect_json_from_value(key: &String, value: OtelValue) -> Map<String, Va
             );
         }
     }
-
-    value_json
 }
 
 /// Recursively converts an ArrayValue into a JSON Value
 /// This handles nested array values and key-value lists by recursively
 /// converting them to JSON
 fn collect_json_from_array_value(array_value: &ArrayValue) -> Value {
-    let mut json_array = Vec::new();
+    let mut json_array = Vec::with_capacity(array_value.values.len());
     for value in &array_value.values {
         if let Some(val) = &value.value {
             match val {
@@ -122,12 +123,11 @@ fn collect_json_from_array_value(array_value: &ArrayValue) -> Value {
 /// The function iterates through the key-value pairs in the list
 /// and collects their JSON representations into a single Map
 fn collect_json_from_key_value_list(key_value_list: KeyValueList) -> Map<String, Value> {
-    let mut kv_list_json: Map<String, Value> = Map::new();
+    let mut kv_list_json: Map<String, Value> = Map::with_capacity(key_value_list.values.len());
     for key_value in key_value_list.values {
         if let Some(val) = key_value.value {
             if let Some(val) = val.value {
-                let json_value = collect_json_from_value(&key_value.key, val);
-                kv_list_json.extend(json_value);
+                insert_json_from_value(&mut kv_list_json, &key_value.key, val);
             } else {
                 tracing::warn!("Key '{}' has no value in key-value list", key_value.key);
             }
@@ -136,37 +136,38 @@ fn collect_json_from_key_value_list(key_value_list: KeyValueList) -> Map<String,
     kv_list_json
 }
 
-pub fn collect_json_from_anyvalue(key: &String, value: AnyValue) -> Map<String, Value> {
+pub fn collect_json_from_anyvalue(key: &str, value: AnyValue) -> Map<String, Value> {
     collect_json_from_value(key, value.value.unwrap())
 }
 
 //traverse through Value by calling function ollect_json_from_any_value
-pub fn collect_json_from_values(values: &Option<AnyValue>, key: &String) -> Map<String, Value> {
-    let mut value_json: Map<String, Value> = Map::new();
+pub fn collect_json_from_values(values: &Option<AnyValue>, key: &str) -> Map<String, Value> {
+    let mut value_json: Map<String, Value> = Map::with_capacity(1);
 
     for value in values.iter() {
-        value_json = collect_json_from_anyvalue(key, value.clone());
+        if let Some(value) = value.value.clone() {
+            insert_json_from_value(&mut value_json, key, value);
+        }
     }
 
     value_json
 }
 
 pub fn value_to_string(value: serde_json::Value) -> String {
-    match value.clone() {
+    match value {
         e @ Value::Number(_) | e @ Value::Bool(_) => e.to_string(),
         Value::String(s) => s,
         _ => "".to_string(),
     }
 }
 
 pub fn flatten_attributes(attributes: &[KeyValue]) -> Map<String, Value> {
-    let mut attributes_json: Map<String, Value> = Map::new();
+    let mut attributes_json: Map<String, Value> = Map::with_capacity(attributes.len());
     for attribute in attributes {
-        let key = &attribute.key;
-        let value = &attribute.value;
-        let value_json = collect_json_from_values(value, &key.to_string());
-        for (attr_key, attr_val) in &value_json {
-            attributes_json.insert(attr_key.clone(), attr_val.clone());
+        if let Some(value) = &attribute.value
+            && let Some(value) = value.value.clone()
+        {
+            insert_json_from_value(&mut attributes_json, &attribute.key, value);
         }
     }
     attributes_json
@@ -193,9 +194,12 @@ pub fn insert_bool_if_some(map: &mut Map<String, Value>, key: &str, option: &Opt
 }
 
 pub fn insert_attributes(map: &mut Map<String, Value>, attributes: &[KeyValue]) {
-    let attributes_json = flatten_attributes(attributes);
-    for (key, value) in attributes_json {
-        map.insert(key, value);
+    for attribute in attributes {
+        if let Some(value) = &attribute.value
+            && let Some(value) = value.value.clone()
+        {
+            insert_json_from_value(map, &attribute.key, value);
+        }
     }
 }
 

src/otel/otel_utils.rs

@@ -41,7 +41,7 @@ use crate::{
 
 use super::StagingError;
 
-const DISK_WRITE_BATCH_ROWS: usize = 16_384;
+const DISK_WRITE_BATCH_ROWS: usize = 32_768;
 
 #[derive(Default)]
 pub struct Writer {
@@ -83,43 +83,97 @@ impl Writer {
             return Ok(());
         }
 
-        let schema = pending.batches[0].schema();
-        let batch = concat_batches(&schema, pending.batches.iter())?;
-        match self.disk.get_mut(filename) {
-            Some(writer) => writer.write(&batch)?,
-            None => {
-                let range = pending.range.expect("pending disk batch must have range");
-                if let Some(parent) = file_path.parent() {
-                    fs::create_dir_all(parent)?;
-                }
-                let mut writer = DiskWriter::try_new(file_path, &schema, range)?;
-                writer.write(&batch)?;
-                self.disk.insert(filename.to_owned(), writer);
-            }
-        }
+        write_pending_disk_batch(&mut self.disk, filename.to_owned(), pending, file_path)?;
 
         Ok(())
     }
 
-    pub fn flush_all_pending_disk(
+    pub fn take_flushable_disk(
         &mut self,
+        forced: bool,
+    ) -> (HashMap<String, DiskWriter>, PendingDiskWrites) {
+        let mut flushable_disk = HashMap::new();
+        let old_disk = std::mem::take(&mut self.disk);
+        for (filename, writer) in old_disk {
+            if !forced && writer.is_current() {
+                self.disk.insert(filename, writer);
+            } else {
+                flushable_disk.insert(filename, writer);
+            }
+        }
+
+        let mut flushable_pending = HashMap::new();
+        let old_pending = std::mem::take(&mut self.disk_pending);
+        for (filename, pending) in old_pending {
+            if !forced && pending.is_current() {
+                self.disk_pending.insert(filename, pending);
+            } else {
+                flushable_pending.insert(filename, pending);
+            }
+        }
+
+        (flushable_disk, PendingDiskWrites(flushable_pending))
+    }
+}
+
+pub struct PendingDiskWrites(HashMap<String, PendingDiskBatch>);
+
+impl PendingDiskWrites {
+    pub fn flush_into(
+        self,
+        disk: &mut HashMap<String, DiskWriter>,
         data_path: &std::path::Path,
     ) -> Result<(), StagingError> {
-        let filenames = self.disk_pending.keys().cloned().collect_vec();
-        for filename in filenames {
-            self.flush_pending_disk(&filename, data_path.join(&filename))?;
+        for (filename, pending) in self.0 {
+            write_pending_disk_batch(disk, filename.clone(), pending, data_path.join(filename))?;
         }
         Ok(())
     }
 }
 
+fn write_pending_disk_batch(
+    disk: &mut HashMap<String, DiskWriter>,
+    filename: String,
+    pending: PendingDiskBatch,
+    file_path: PathBuf,
+) -> Result<(), StagingError> {
+    if pending.batches.is_empty() {
+        return Ok(());
+    }
+
+    let schema = pending.batches[0].schema();
+    let batch = concat_batches(&schema, pending.batches.iter())?;
+    match disk.get_mut(&filename) {
+        Some(writer) => writer.write(&batch)?,
+        None => {
+            let range = pending.range.expect("pending disk batch must have range");
+            if let Some(parent) = file_path.parent() {
+                fs::create_dir_all(parent)?;
+            }
+            let mut writer = DiskWriter::try_new(file_path, &schema, range)?;
+            writer.write(&batch)?;
+            disk.insert(filename, writer);
+        }
+    }
+
+    Ok(())
+}
+
 #[derive(Default)]
 struct PendingDiskBatch {
     rows: usize,
     batches: Vec<RecordBatch>,
     range: Option<TimeRange>,
 }
 
+impl PendingDiskBatch {
+    fn is_current(&self) -> bool {
+        self.range
+            .as_ref()
+            .is_some_and(|range| range.contains(Utc::now()))
+    }
+}
+
 pub struct DiskWriter {
     inner: StreamWriter<BufWriter<File>>,
     path: PathBuf,