Hottier fix

src/cli.rs

@@ -314,6 +314,50 @@ 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",
+        default_value = "4",
+        help = "Number of concurrent parquet file downloads per stream during hot tier sync"
+    )]
+    pub hot_tier_files_per_stream_concurrency: usize,
+
+    #[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-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",

src/handlers/http/logstream.rs

@@ -413,13 +413,21 @@ 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);
+    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
@@ -469,19 +477,29 @@ pub async fn put_stream_hot_tier(
         .metastore
         .put_stream_json(&stream_metadata, &stream_name, &tenant_id)
         .await?;
-
+    hot_tier_manager
+        .spawn_stream_tasks(stream_name.clone(), tenant_id.clone())
+        .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
@@ -502,12 +520,20 @@ 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);
+    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

src/handlers/http/modal/mod.rs

@@ -160,6 +160,11 @@ pub trait ParseableServer {
             // Shutdown resource monitor
             let _ = resource_shutdown_tx.send(());
 
+            // Shutdown hottier
+            if let Some(ht_global) = HotTierManager::global() {
+                ht_global.abort_all().await;
+            }
+
             // Initiate graceful shutdown
             info!("Graceful shutdown of HTTP server triggered");
             srv_handle.stop(true).await;
@@ -626,6 +631,7 @@ pub type PrismMetadata = NodeMetadata;
 /// Initialize hot tier metadata files for streams that have hot tier configuration
 /// in their stream metadata but don't have local hot tier metadata files yet.
 /// This function is called once during query server startup.
+#[tracing::instrument(name = "hottier.init_metadata_startup", skip(hot_tier_manager))]
 pub async fn initialize_hot_tier_metadata_on_startup(
     hot_tier_manager: &HotTierManager,
 ) -> anyhow::Result<()> {