Merge branch 'main' into helm-2.8.0

Signed-off-by: Nitish Tiwari <nitish@parseable.com>

Author: nitisht

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "parseable"
-version = "2.7.2"
+version = "2.8.0"
 authors = ["Parseable Team <hi@parseable.com>"]
 edition = "2024"
 rust-version = "1.88.0"
@@ -127,6 +127,7 @@ opentelemetry-otlp = { version = "0.31.1", features = [
     "http-proto",
     "http-json",
 ] }
+opentelemetry-stdout = "0.31.0"
 tracing-actix-web = "0.7"
 
 # Time and Date
@@ -180,6 +181,7 @@ thiserror = "2.0"
 ulid = { version = "1.0", features = ["serde"] }
 uuid = { version = "1", features = ["v4"] }
 xxhash-rust = { version = "0.8", features = ["xxh3"] }
+rustc-hash = "2"
 futures-core = "0.3.31"
 tempfile = "3.20.0"
 lazy_static = "1.4.0"
@@ -204,8 +206,8 @@ arrow = "58.0.0"
 temp-dir = "0.1.14"
 
 [package.metadata.parseable_ui]
-assets-url = "https://parseable-prism-build.s3.us-east-2.amazonaws.com/v2.7.2/build.zip"
-assets-sha1 = "541746bd89b41528477a57a2f6f066dce145dd73"
+assets-url = "https://parseable-prism-build.s3.us-east-2.amazonaws.com/v2.8.0/build.zip"
+assets-sha1 = "a7523ef819d38678275ae165c443564b2f9a3fc1"
 
 [features]
 debug = []

Cargo.toml

@@ -17,7 +17,7 @@
  */
 
 use clap::{Parser, value_parser};
-use std::{env, fs, path::PathBuf};
+use std::{env, fs, ops::Div, path::PathBuf};
 
 use url::Url;
 
@@ -194,6 +194,16 @@ pub struct Options {
     )]
     pub max_connections: usize,
 
+    // DataFusion target partitions
+    #[arg(
+        long,
+        env = "P_DATAFUSION_TARGET_PARTITIONS",
+        default_value_t = num_cpus::get().div(2).max(1) as u64,
+        value_parser = value_parser!(u64).range(1..),
+        help = "Number of partitions for DF to split execution into"
+    )]
+    pub target_partitions: u64,
+
     #[arg(
         long = "origin",
         env = "P_ORIGIN_URI",
@@ -314,6 +324,60 @@ pub struct Options {
     )]
     pub hot_tier_storage_path: Option<PathBuf>,
 
+    #[arg(
+        long = "hot-tier-download-chunk-size",
+        env = "P_HOT_TIER_DOWNLOAD_CHUNK_SIZE",
+        value_parser = clap::value_parser!(u64).range(5242880..),
+        default_value = "8388608",
+        help = "Chunk size in bytes for parallel hot tier downloads (default 8 MiB)"
+    )]
+    pub hot_tier_download_chunk_size: u64,
+
+    #[arg(
+        long = "hot-tier-download-concurrency",
+        env = "P_HOT_TIER_DOWNLOAD_CONCURRENCY",
+        value_parser = clap::value_parser!(u64).range(1..),
+        default_value = "16",
+        help = "Number of concurrent range requests per hot tier download"
+    )]
+    pub hot_tier_download_concurrency: u64,
+
+    #[arg(
+        long = "hot-tier-files-per-stream-concurrency",
+        env = "P_HOT_TIER_FILES_PER_STREAM_CONCURRENCY",
+        value_parser = clap::value_parser!(u32).range(1..),
+        default_value = "4",
+        help = "Number of concurrent parquet file downloads per stream during hot tier sync"
+    )]
+    pub hot_tier_files_per_stream_concurrency: u32,
+
+    #[arg(
+        long = "hot-tier-latest-minutes",
+        env = "P_HOT_TIER_LATEST_MINUTES",
+        value_parser = clap::value_parser!(u64).range(1..),
+        default_value = "10",
+        help = "Files whose timestamp is within the last N minutes are 'latest'; rest are 'historic'."
+    )]
+    pub hot_tier_latest_minutes: u64,
+
+    #[arg(
+        long = "hot-tier-per-tick-cap",
+        env = "P_HISTORIC_PER_TICK_CAP",
+        value_parser = clap::value_parser!(u32).range(10..),
+        default_value = "100",
+        help = "Maximum files to download per historic tick."
+    )]
+    pub historic_per_tick_cap: u32,
+
+    #[arg(
+        long = "hot-tier-historic-sync-minutes",
+        env = "P_HOT_TIER_HISTORIC_SYNC_MINUTES",
+        value_parser = clap::value_parser!(u32).range(1..),
+        default_value = "5",
+        help = "Interval (minutes) at which the historic hot-tier sync runs."
+    )]
+    pub hot_tier_historic_sync_minutes: u32,
+
     //TODO: remove this when smart cache is implemented
     #[arg(
         long = "index-storage-path",

helm-releases/parseable-enterprise-2.7.3.tgz

@@ -235,7 +235,7 @@ impl FlightService for AirServiceImpl {
 
         if event.is_some() {
             // Clear staging of stream once airplane has taxied
-            PARSEABLE.get_or_create_stream(&stream_name, &None).clear();
+            let _ = PARSEABLE.get_or_create_stream(&stream_name, &None).clear();
         }
 
         let time = time.elapsed().as_secs_f64();

src/cli.rs

@@ -20,7 +20,7 @@ use self::error::StreamError;
 use super::cluster::utils::{IngestionStats, QueriedStats, StorageStats};
 use super::query::update_schema_when_distributed;
 use crate::event::format::override_data_type;
-use crate::hottier::{CURRENT_HOT_TIER_VERSION, HotTierManager, StreamHotTier};
+use crate::hottier::{CURRENT_HOT_TIER_VERSION, GLOBAL_HOTTIER, StreamHotTier};
 use crate::metadata::SchemaVersion;
 use crate::metrics::{EVENTS_INGESTED_DATE, EVENTS_INGESTED_SIZE_DATE, EVENTS_STORAGE_SIZE_DATE};
 use crate::parseable::{DEFAULT_TENANT, PARSEABLE, StreamNotFound};
@@ -47,7 +47,7 @@ use itertools::Itertools;
 use serde_json::{Value, json};
 use std::fs;
 use std::sync::Arc;
-use tracing::warn;
+use tracing::{Instrument, warn};
 
 pub async fn delete(
     req: HttpRequest,
@@ -77,7 +77,7 @@ pub async fn delete(
         )
     }
 
-    if let Some(hot_tier_manager) = HotTierManager::global()
+    if let Some(hot_tier_manager) = GLOBAL_HOTTIER.get()
         && hot_tier_manager.check_stream_hot_tier_exists(&stream_name, &tenant_id)
     {
         hot_tier_manager
@@ -413,16 +413,25 @@ pub async fn get_stream_info(
     Ok((web::Json(stream_info), StatusCode::OK))
 }
 
+#[tracing::instrument(
+    name = "http.put_stream_hot_tier",
+    skip(req, logstream, hottier),
+    fields(stream = tracing::field::Empty, tenant = tracing::field::Empty, size = hottier.size)
+)]
 pub async fn put_stream_hot_tier(
     req: HttpRequest,
     logstream: Path<String>,
     Json(mut hottier): Json<StreamHotTier>,
 ) -> Result<impl Responder, StreamError> {
     let stream_name = logstream.into_inner();
     let tenant_id = get_tenant_id_from_request(&req);
+    let current_span = tracing::Span::current();
+    current_span
+        .record("stream", tracing::field::display(&stream_name))
+        .record("tenant", tracing::field::debug(&tenant_id));
     // For query mode, if the stream not found in memory map,
-    //check if it exists in the storage
-    //create stream and schema from storage
+    // check if it exists in the storage
+    // create stream and schema from storage
     if !PARSEABLE
         .check_or_load_stream(&stream_name, &tenant_id)
         .await
@@ -441,16 +450,14 @@ pub async fn put_stream_hot_tier(
 
     validator::hot_tier(&hottier.size.to_string())?;
 
-    // TODO tenants
-    stream.set_hot_tier(Some(hottier.clone()));
-    let Some(hot_tier_manager) = HotTierManager::global() else {
+    let Some(hot_tier_manager) = GLOBAL_HOTTIER.get() else {
         return Err(StreamError::HotTierNotEnabled(stream_name));
     };
     let existing_hot_tier_used_size = hot_tier_manager
         .validate_hot_tier_size(&stream_name, hottier.size, &tenant_id)
         .await?;
     hottier.used_size = existing_hot_tier_used_size;
-    hottier.available_size = hottier.size;
+    hottier.available_size = hottier.size.saturating_sub(existing_hot_tier_used_size);
     hottier.version = Some(CURRENT_HOT_TIER_VERSION.to_string());
     hot_tier_manager
         .put_hot_tier(&stream_name, &mut hottier, &tenant_id)
@@ -469,19 +476,34 @@ pub async fn put_stream_hot_tier(
         .metastore
         .put_stream_json(&stream_metadata, &stream_name, &tenant_id)
         .await?;
+    stream.set_hot_tier(Some(hottier.clone()));
+    let stream = stream_name.clone();
+    let tenant = tenant_id.clone();
+    hot_tier_manager
+        .spawn_stream_task(stream, tenant)
+        .instrument(current_span)
+        .await;
 
     Ok((
         format!("hot tier set for stream {stream_name}"),
         StatusCode::OK,
     ))
 }
 
+#[tracing::instrument(
+    name = "http.get_stream_hot_tier",
+    skip(req, logstream),
+    fields(stream = tracing::field::Empty, tenant = tracing::field::Empty)
+)]
 pub async fn get_stream_hot_tier(
     req: HttpRequest,
     logstream: Path<String>,
 ) -> Result<impl Responder, StreamError> {
     let stream_name = logstream.into_inner();
     let tenant_id = get_tenant_id_from_request(&req);
+    tracing::Span::current()
+        .record("stream", tracing::field::display(&stream_name))
+        .record("tenant", tracing::field::debug(&tenant_id));
     // For query mode, if the stream not found in memory map,
     //check if it exists in the storage
     //create stream and schema from storage
@@ -492,7 +514,7 @@ pub async fn get_stream_hot_tier(
         return Err(StreamNotFound(stream_name.clone()).into());
     }
 
-    let Some(hot_tier_manager) = HotTierManager::global() else {
+    let Some(hot_tier_manager) = GLOBAL_HOTTIER.get() else {
         return Err(StreamError::HotTierNotEnabled(stream_name));
     };
     let meta = hot_tier_manager
@@ -502,12 +524,21 @@ pub async fn get_stream_hot_tier(
     Ok((web::Json(meta), StatusCode::OK))
 }
 
+#[tracing::instrument(
+    name = "http.delete_stream_hot_tier",
+    skip(req, logstream),
+    fields(stream = tracing::field::Empty, tenant = tracing::field::Empty)
+)]
 pub async fn delete_stream_hot_tier(
     req: HttpRequest,
     logstream: Path<String>,
 ) -> Result<impl Responder, StreamError> {
     let stream_name = logstream.into_inner();
     let tenant_id = get_tenant_id_from_request(&req);
+    let current_span = tracing::Span::current();
+    current_span
+        .record("stream", tracing::field::display(&stream_name))
+        .record("tenant", tracing::field::debug(&tenant_id));
     // For query mode, if the stream not found in memory map,
     //check if it exists in the storage
     //create stream and schema from storage
@@ -529,12 +560,13 @@ pub async fn delete_stream_hot_tier(
         });
     }
 
-    let Some(hot_tier_manager) = HotTierManager::global() else {
+    let Some(hot_tier_manager) = GLOBAL_HOTTIER.get() else {
         return Err(StreamError::HotTierNotEnabled(stream_name));
     };
 
     hot_tier_manager
         .delete_hot_tier(&stream_name, &tenant_id)
+        .instrument(tracing::Span::current())
         .await?;
 
     let mut stream_metadata: ObjectStoreFormat = serde_json::from_slice(

src/handlers/airplane.rs

@@ -38,7 +38,7 @@ use crate::{
     alerts::{ALERTS, get_alert_manager, target::TARGETS},
     cli::Options,
     correlation::CORRELATIONS,
-    hottier::{HotTierManager, StreamHotTier},
+    hottier::{GLOBAL_HOTTIER, HotTierManager, StreamHotTier},
     metastore::metastore_traits::MetastoreObject,
     oauth::{OAuthProvider, connect_oidc},
     option::Mode,
@@ -160,6 +160,11 @@ pub trait ParseableServer {
             // Shutdown resource monitor
             let _ = resource_shutdown_tx.send(());
 
+            // Shutdown hottier
+            if let Some(htm) = GLOBAL_HOTTIER.get() {
+                htm.abort_all().await;
+            }
+
             // Initiate graceful shutdown
             info!("Graceful shutdown of HTTP server triggered");
             srv_handle.stop(true).await;
@@ -627,7 +632,7 @@ pub type PrismMetadata = NodeMetadata;
 /// in their stream metadata but don't have local hot tier metadata files yet.
 /// This function is called once during query server startup.
 pub async fn initialize_hot_tier_metadata_on_startup(
-    hot_tier_manager: &HotTierManager,
+    hot_tier_manager: &'static HotTierManager,
 ) -> anyhow::Result<()> {
     // Collect hot tier configurations from streams before doing async operations
     let hot_tier_configs: Vec<(String, Option<String>, StreamHotTier)> = {
@@ -653,21 +658,6 @@ pub async fn initialize_hot_tier_metadata_on_startup(
                 })
             })
             .collect()
-        // let streams_guard = PARSEABLE.streams.read().unwrap();
-        // streams_guard
-        //     .iter()
-        //     .filter_map(|(stream_name, stream)| {
-        //         // Skip if hot tier metadata file already exists for this stream
-        //         if hot_tier_manager.check_stream_hot_tier_exists(stream_name) {
-        //             return None;
-        //         }
-
-        //         // Get the hot tier configuration from the in-memory stream metadata
-        //         stream
-        //             .get_hot_tier()
-        //             .map(|config| (stream_name.clone(), config))
-        //     })
-        //     .collect()
     };
 
     for (stream_name, tenant_id, hot_tier_config) in hot_tier_configs {