add conversion and upload duration to histogram metric (#1793)

* add conversion and upload duration to histogram metric

metrics added - conversion_task_duration, object_store_sync_duration
with phase - periodic and startup

remove conversion and object store sync tasks from query server

* custom buckets for conversion and query execution time

Author: nikhilsinhaparseable

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::option::Mode;
 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};
@@ -65,11 +66,13 @@ pub async fn shutdown() {
     //sleep for 5 secs to allow any ongoing requests to finish
     tokio::time::sleep(std::time::Duration::from_secs(5)).await;
 
-    // Perform sync operations
-    perform_sync_operations().await;
+    // Only ingestion-capable nodes own local data that needs flushing.
+    if matches!(PARSEABLE.options.mode, Mode::Ingest | Mode::All) {
+        perform_sync_operations().await;
 
-    // This is to ensure that all stats data is synced before the server shuts down
-    perform_sync_operations().await;
+        // This is to ensure that all stats data is synced before the server shuts down
+        perform_sync_operations().await;
+    }
 }
 
 async fn set_shutdown_flag() {

src/handlers/http/modal/query_server.rs

@@ -17,7 +17,6 @@
  */
 
 use std::sync::Arc;
-use std::thread;
 
 use crate::handlers::airplane;
 use crate::handlers::http::cluster;
@@ -33,7 +32,7 @@ use crate::hottier::HotTierManager;
 use crate::hottier::HotTierMessage;
 use crate::hottier::hottier_runtime;
 use crate::rbac::role::Action;
-use crate::{analytics, migration, storage, sync};
+use crate::{analytics, migration, storage};
 use actix_web::web::{ServiceConfig, resource};
 use actix_web::{Scope, web};
 use actix_web_prometheus::PrometheusMetrics;
@@ -159,18 +158,10 @@ impl ParseableServer for QueryServer {
             htm.start_all_tasks().await;
         };
 
-        // Run sync on a background thread
-        let (cancel_tx, cancel_rx) = oneshot::channel();
-        thread::spawn(|| sync::handler(cancel_rx));
-
         tokio::spawn(airplane::server());
 
-        let result = self
-            .start(shutdown_rx, prometheus.clone(), PARSEABLE.options.openid())
-            .await?;
-        // Cancel sync jobs
-        cancel_tx.send(()).expect("Cancellation should not fail");
-        Ok(result)
+        self.start(shutdown_rx, prometheus.clone(), PARSEABLE.options.openid())
+            .await
     }
 }
 

src/metrics/mod.rs

@@ -35,6 +35,14 @@ use prometheus::{
 
 pub const METRICS_NAMESPACE: &str = env!("CARGO_PKG_NAME");
 
+const SYNC_TASK_DURATION_BUCKETS: &[f64] = &[
+    15.0, 30.0, 45.0, 60.0, 90.0, 120.0, 150.0, 180.0, 300.0, 600.0,
+];
+
+const QUERY_EXECUTE_TIME_BUCKETS: &[f64] = &[
+    0.1, 0.2, 0.5, 1.0, 5.0, 15.0, 30.0, 45.0, 60.0, 90.0, 120.0, 150.0, 180.0,
+];
+
 pub static METRICS_REGISTRY: Lazy<Registry> = Lazy::new(|| {
     let registry = Registry::new();
     custom_metrics(&registry);
@@ -322,12 +330,40 @@ mod process_metrics_tests {
 
 pub static QUERY_EXECUTE_TIME: Lazy<HistogramVec> = Lazy::new(|| {
     HistogramVec::new(
-        HistogramOpts::new("query_execute_time", "Query execute time").namespace(METRICS_NAMESPACE),
+        HistogramOpts::new("query_execute_time", "Query execute time")
+            .namespace(METRICS_NAMESPACE)
+            .buckets(QUERY_EXECUTE_TIME_BUCKETS.to_vec()),
         &["stream", "tenant_id"],
     )
     .expect("metric can be created")
 });
 
+pub static CONVERSION_TASK_DURATION: Lazy<HistogramVec> = Lazy::new(|| {
+    HistogramVec::new(
+        HistogramOpts::new(
+            "conversion_task_duration",
+            "Local conversion task duration in seconds",
+        )
+        .namespace(METRICS_NAMESPACE)
+        .buckets(SYNC_TASK_DURATION_BUCKETS.to_vec()),
+        &["phase"],
+    )
+    .expect("metric can be created")
+});
+
+pub static OBJECT_STORE_SYNC_DURATION: Lazy<HistogramVec> = Lazy::new(|| {
+    HistogramVec::new(
+        HistogramOpts::new(
+            "object_store_sync_duration",
+            "Object store sync duration in seconds",
+        )
+        .namespace(METRICS_NAMESPACE)
+        .buckets(SYNC_TASK_DURATION_BUCKETS.to_vec()),
+        &["phase"],
+    )
+    .expect("metric can be created")
+});
+
 pub static QUERY_CACHE_HIT: Lazy<IntCounterVec> = Lazy::new(|| {
     IntCounterVec::new(
         Opts::new("QUERY_CACHE_HIT", "Full Cache hit").namespace(METRICS_NAMESPACE),
@@ -838,6 +874,12 @@ fn custom_metrics(registry: &Registry) {
     registry
         .register(Box::new(QUERY_EXECUTE_TIME.clone()))
         .expect("metric can be registered");
+    registry
+        .register(Box::new(CONVERSION_TASK_DURATION.clone()))
+        .expect("metric can be registered");
+    registry
+        .register(Box::new(OBJECT_STORE_SYNC_DURATION.clone()))
+        .expect("metric can be registered");
     registry
         .register(Box::new(QUERY_CACHE_HIT.clone()))
         .expect("metric can be registered");

src/sync.rs

@@ -82,6 +82,7 @@ pub async fn shutdown_local_sync_flush_and_convert() {
 
 use crate::alerts::alert_enums::AlertTask;
 use crate::alerts::alerts_utils;
+use crate::metrics::{CONVERSION_TASK_DURATION, OBJECT_STORE_SYNC_DURATION};
 use crate::parseable::PARSEABLE;
 use crate::storage::object_storage::sync_all_streams;
 use crate::{LOCAL_SYNC_INTERVAL, STORAGE_UPLOAD_INTERVAL};
@@ -98,7 +99,13 @@ fn next_minute() -> Instant {
     Instant::now() + time_till.to_std().expect("Valid duration")
 }
 
-pub async fn monitor_task_duration<F, Fut, T>(task_name: &str, threshold: Duration, f: F) -> T
+pub async fn monitor_task_duration<F, Fut, T>(
+    task_name: &str,
+    threshold: Duration,
+    duration_metric: &prometheus::HistogramVec,
+    phase: &str,
+    f: F,
+) -> T
 where
     F: FnOnce() -> Fut + Send + 'static,
     Fut: Future<Output = T> + Send,
@@ -117,10 +124,14 @@ where
                 warned_once = true;
             },
             res = &mut future => {
+                let elapsed = start_time.elapsed();
+                duration_metric
+                    .with_label_values(&[phase])
+                    .observe(elapsed.as_secs_f64());
                 if warned_once {
                     warn!(
                         "Task '{task_name}' took longer than expected: {:?} (threshold: {threshold:?})",
-                        start_time.elapsed()
+                        elapsed
                     );
                 }
                 break res.expect("Task handle shouldn't error");
@@ -191,6 +202,8 @@ pub fn object_store_sync() -> (
                             monitor_task_duration(
                                 "object_store_sync_all_streams",
                                 Duration::from_secs(PARSEABLE.options.object_store_sync_threshold),
+                                &OBJECT_STORE_SYNC_DURATION,
+                                "periodic",
                                 || async {
                                     let mut joinset = JoinSet::new();
                                     sync_all_streams(&mut joinset);
@@ -312,6 +325,8 @@ async fn run_periodic_local_sync_cycle(_permit: OwnedSemaphorePermit) {
         monitor_task_duration(
             "local_sync_flush_and_convert",
             Duration::from_secs(PARSEABLE.options.local_sync_threshold),
+            &CONVERSION_TASK_DURATION,
+            "periodic",
             || async {
                 let mut joinset = JoinSet::new();
                 PARSEABLE
@@ -357,6 +372,8 @@ async fn sync_start_inner(
         monitor_task_duration(
             "startup_local_sync",
             Duration::from_secs(PARSEABLE.options.local_sync_threshold),
+            &CONVERSION_TASK_DURATION,
+            "startup",
             || async move {
                 // Plans contain all paths from the startup snapshot. Execute
                 // streams and their event-minute groups sequentially to keep
@@ -386,6 +403,8 @@ async fn sync_start_inner(
         monitor_task_duration(
             "startup_object_store_sync",
             Duration::from_secs(PARSEABLE.options.object_store_sync_threshold),
+            &OBJECT_STORE_SYNC_DURATION,
+            "startup",
             || async {
                 let mut object_store_joinset = JoinSet::new();
                 sync_all_streams(&mut object_store_joinset);