add traces to hottier

Author: parmesant

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
@@ -478,12 +486,20 @@ pub async fn put_stream_hot_tier(
     ))
 }
 
+#[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
@@ -504,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

@@ -626,6 +626,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<()> {

src/hottier.rs

@@ -144,6 +144,11 @@ impl HotTierManager {
     /// deletes parquet files that exist but are not in their date manifest,
     /// then recomputes `used_size` / `available_size` from the cleaned
     /// manifests and persists the updated `StreamHotTier`.
+    #[tracing::instrument(
+        name = "hottier.reconcile_stream",
+        skip(self),
+        fields(stream = %stream, tenant = ?tenant_id)
+    )]
     async fn reconcile_stream(
         &self,
         stream: &str,
@@ -217,6 +222,11 @@ impl HotTierManager {
         Ok(sht)
     }
 
+    #[tracing::instrument(
+        name = "hottier.drop_partials",
+        skip(self, on_disk, partials_removed),
+        fields(stream = %stream, tenant = ?tenant_id, date_dir = %date_dir.display())
+    )]
     async fn drop_partials(
         &self,
         on_disk: &mut HashSet<String>,
@@ -264,6 +274,11 @@ impl HotTierManager {
         Ok(())
     }
 
+    #[tracing::instrument(
+        name = "hottier.clean_manifest",
+        skip(self, keep_names, total_used, entries_dropped),
+        fields(stream = %stream, tenant = ?tenant_id, date_dir = %date_dir.display())
+    )]
     async fn clean_manifest(
         &self,
         keep_names: &mut HashSet<String>,
@@ -327,6 +342,11 @@ impl HotTierManager {
     }
 
     /// get the total hot tier size for all streams
+    #[tracing::instrument(
+        name = "hottier.get_hot_tiers_size",
+        skip(self),
+        fields(current_stream = %current_stream, current_tenant = ?current_tenant_id)
+    )]
     pub async fn get_hot_tiers_size(
         &self,
         current_stream: &str,
@@ -358,6 +378,11 @@ impl HotTierManager {
     /// check disk usage and hot tier size of all other streams
     /// check if total hot tier size of all streams is less than max disk usage
     /// delete all the files from hot tier once validation is successful and hot tier is ready to be updated
+    #[tracing::instrument(
+        name = "hottier.validate_size",
+        skip(self),
+        fields(stream = %stream, tenant = ?tenant_id, size = stream_hot_tier_size)
+    )]
     pub async fn validate_hot_tier_size(
         &self,
         stream: &str,
@@ -420,6 +445,11 @@ impl HotTierManager {
     }
 
     /// get the hot tier metadata file for the stream
+    #[tracing::instrument(
+        name = "hottier.get_hot_tier",
+        skip(self),
+        fields(stream = %stream, tenant = ?tenant_id)
+    )]
     pub async fn get_hot_tier(
         &self,
         stream: &str,
@@ -442,6 +472,11 @@ impl HotTierManager {
         Ok(stream_hot_tier)
     }
 
+    #[tracing::instrument(
+        name = "hottier.delete_hot_tier",
+        skip(self),
+        fields(stream = %stream, tenant = ?tenant_id)
+    )]
     pub async fn delete_hot_tier(
         &self,
         stream: &str,
@@ -466,6 +501,11 @@ impl HotTierManager {
 
     /// put the hot tier metadata file for the stream
     /// set the updated_date_range in the hot tier metadata file
+    #[tracing::instrument(
+        name = "hottier.put_hot_tier",
+        skip(self, hot_tier),
+        fields(stream = %stream, tenant = ?tenant_id, size = hot_tier.size)
+    )]
     pub async fn put_hot_tier(
         &self,
         stream: &str,
@@ -502,6 +542,7 @@ impl HotTierManager {
     /// Discover hot-tier-enabled streams at boot and spawn a per-stream pair
     /// of (Latest, Historic) loops for each. New streams added later acquire
     /// their own loops via `spawn_stream_tasks` from the PUT hot-tier handler.
+    #[tracing::instrument(name = "hottier.startup", skip(self))]
     pub fn download_from_s3<'a>(&'a self) -> Result<(), HotTierError>
     where
         'a: 'static,
@@ -554,6 +595,11 @@ impl HotTierManager {
 
     /// Spawn (Latest, Historic) loops for a single stream. Idempotent:
     /// if tasks already exist for this (tenant, stream), no-op.
+    #[tracing::instrument(
+        name = "hottier.spawn_stream_tasks",
+        skip(self),
+        fields(stream = %stream, tenant = ?tenant_id)
+    )]
     pub async fn spawn_stream_tasks(&'static self, stream: String, tenant_id: Option<String>) {
         let key: StreamKey = (tenant_id.clone(), stream.clone());
         {
@@ -628,6 +674,11 @@ impl HotTierManager {
 
     /// process the hot tier files for the stream
     /// delete the files from the hot tier directory if the available date range is outside the hot tier range
+    #[tracing::instrument(
+        name = "hottier.process_stream",
+        skip(self),
+        fields(stream = %stream, tenant = ?tenant_id, phase = ?phase)
+    )]
     async fn process_stream(
         &self,
         stream: String,
@@ -681,6 +732,11 @@ impl HotTierManager {
     /// collect all manifests from metastore for the date, sort the parquet file list
     /// in order to download the latest files first
     /// download the parquet files if not present in hot tier directory
+    #[tracing::instrument(
+        name = "hottier.process_manifest",
+        skip(self, manifest_files_to_download),
+        fields(stream = %stream, tenant = ?tenant_id, phase = ?phase, dates = manifest_files_to_download.len())
+    )]
     async fn process_manifest(
         &self,
         stream: &str,
@@ -809,6 +865,18 @@ impl HotTierManager {
     /// Returns false when no budget is available (caller should stop scheduling
     /// further work for this stream).
     #[allow(clippy::too_many_arguments)]
+    #[tracing::instrument(
+        name = "hottier.process_parquet_file",
+        skip(self, parquet_file, parquet_path, state),
+        fields(
+            stream = %stream,
+            tenant = ?tenant_id,
+            phase = ?phase,
+            date = %date,
+            file = %parquet_file.file_path,
+            file_size = parquet_file.file_size
+        )
+    )]
     async fn process_parquet_file_concurrent(
         &self,
         stream: &str,
@@ -999,6 +1067,11 @@ impl HotTierManager {
     }
 
     ///fetch the list of dates available in the hot tier directory for the stream and sort them
+    #[tracing::instrument(
+        name = "hottier.fetch_dates",
+        skip(self),
+        fields(stream = %stream, tenant = ?tenant_id)
+    )]
     pub async fn fetch_hot_tier_dates(
         &self,
         stream: &str,
@@ -1112,6 +1185,11 @@ impl HotTierManager {
     }
 
     ///get the list of parquet files from the hot tier directory for the stream
+    #[tracing::instrument(
+        name = "hottier.get_parquet_files",
+        skip(self),
+        fields(stream = %stream, tenant = ?tenant_id)
+    )]
     pub async fn get_hot_tier_parquet_files(
         &self,
         stream: &str,
@@ -1164,6 +1242,11 @@ impl HotTierManager {
     /// check for the oldest entry to delete if the path exists in hot tier
     /// update the used and available size in the hot tier metadata
     /// loop if available size is still less than the parquet file size
+    #[tracing::instrument(
+        name = "hottier.cleanup_old_data",
+        skip(self, stream_hot_tier, download_file_path),
+        fields(stream = %stream, tenant = ?tenant_id, target_size = parquet_file_size)
+    )]
     pub async fn cleanup_hot_tier_old_data(
         &self,
         stream: &str,
@@ -1408,6 +1491,7 @@ impl HotTierManager {
         Ok(None)
     }
 
+    #[tracing::instrument(name = "hottier.put_internal_stream", skip(self))]
     pub async fn put_internal_stream_hot_tier(&self) -> Result<(), HotTierError> {
         let tenants = if let Some(tenants) = PARSEABLE.list_tenants() {
             tenants.into_iter().map(Some).collect()
@@ -1439,6 +1523,7 @@ impl HotTierManager {
     }
 
     /// Creates hot tier for pstats internal stream if the stream exists in storage
+    #[tracing::instrument(name = "hottier.create_pstats", skip(self))]
     async fn create_pstats_hot_tier(&self) -> Result<(), HotTierError> {
         let tenants = if let Some(tenants) = PARSEABLE.list_tenants() {
             tenants.into_iter().map(Some).collect()

src/metrics/mod.rs

@@ -389,6 +389,18 @@ pub static STORAGE_REQUEST_RESPONSE_TIME: Lazy<HistogramVec> = Lazy::new(|| {
     .expect("metric can be created")
 });
 
+pub static STORAGE_REQUESTS_INFLIGHT: Lazy<IntGaugeVec> = Lazy::new(|| {
+    IntGaugeVec::new(
+        Opts::new(
+            "storage_requests_inflight",
+            "Number of in-flight object store requests",
+        )
+        .namespace(METRICS_NAMESPACE),
+        &["provider", "method"],
+    )
+    .expect("metric can be created")
+});
+
 pub static TOTAL_METRICS_COLLECTED_BY_DATE: Lazy<IntCounterVec> = Lazy::new(|| {
     IntCounterVec::new(
         Opts::new(
@@ -565,6 +577,9 @@ fn custom_metrics(registry: &Registry) {
     registry
         .register(Box::new(STORAGE_REQUEST_RESPONSE_TIME.clone()))
         .expect("metric can be registered");
+    registry
+        .register(Box::new(STORAGE_REQUESTS_INFLIGHT.clone()))
+        .expect("metric can be registered");
     registry
         .register(Box::new(TOTAL_METRICS_COLLECTED_BY_DATE.clone()))
         .expect("metric can be registered");