refactor: remove slow query metric and expose query buckets (#1799)

Author: pratik50

src/handlers/http/query.rs

@@ -46,9 +46,7 @@ use tokio::task::JoinSet;
 use tracing::{error, warn};
 
 use crate::event::{DEFAULT_TIMESTAMP_KEY, commit_schema};
-use crate::metrics::{
-    QUERY_EXECUTE_TIME, QueryType, increment_query_calls_by_date, record_query_metrics,
-};
+use crate::metrics::{QUERY_EXECUTE_TIME, increment_query_calls_by_date};
 use crate::parseable::{DEFAULT_TENANT, PARSEABLE, StreamNotFound};
 use crate::query::error::ExecuteError;
 use crate::query::resolve_stream_names;
@@ -160,8 +158,6 @@ pub async fn query(req: HttpRequest, query_request: Query) -> Result<HttpRespons
     let mut session_state = QUERY_SESSION.get_ctx().state();
     let time_range =
         TimeRange::parse_human_time(&query_request.start_time, &query_request.end_time)?;
-    let query_range_seconds =
-        (time_range.end - time_range.start).num_milliseconds() as f64 / 1_000.0;
     let tables = resolve_stream_names(&query_request.query)?;
     // check or load streams in memory
     create_streams_for_distributed(tables.clone(), &get_tenant_id_from_request(&req)).await?;
@@ -194,29 +190,13 @@ pub async fn query(req: HttpRequest, query_request: Query) -> Result<HttpRespons
         let table = tables
             .first()
             .ok_or_else(|| QueryError::MalformedQuery("No table name found in query"))?;
-        return handle_count_query(
-            &query_request,
-            table,
-            column_name,
-            time,
-            query_range_seconds,
-            &tenant_id,
-        )
-        .await;
+        return handle_count_query(&query_request, table, column_name, time, &tenant_id).await;
     }
 
     // if the query request has streaming = false (default)
     // we use datafusion's `execute` method to get the records
     if !query_request.streaming {
-        return handle_non_streaming_query(
-            query,
-            tables,
-            &query_request,
-            time,
-            query_range_seconds,
-            &tenant_id,
-        )
-        .await;
+        return handle_non_streaming_query(query, tables, &query_request, time, &tenant_id).await;
     }
 
     // if the query request has streaming = true
@@ -243,7 +223,6 @@ async fn handle_count_query(
     table_name: &str,
     column_name: &str,
     time: Instant,
-    query_range_seconds: f64,
     tenant_id: &Option<String>,
 ) -> Result<HttpResponse, QueryError> {
     let counts_req = CountsRequest {
@@ -270,7 +249,6 @@ async fn handle_count_query(
     QUERY_EXECUTE_TIME
         .with_label_values(&[table_name, tenant_id.as_deref().unwrap_or(DEFAULT_TENANT)])
         .observe(time);
-    record_query_metrics(time, query_range_seconds, QueryType::Sql);
 
     Ok(HttpResponse::Ok()
         .insert_header((TIME_ELAPSED_HEADER, total_time.as_str()))
@@ -296,7 +274,6 @@ async fn handle_non_streaming_query(
     table_name: Vec<String>,
     query_request: &Query,
     time: Instant,
-    query_range_seconds: f64,
     tenant_id: &Option<String>,
 ) -> Result<HttpResponse, QueryError> {
     let first_table_name = table_name[0].clone();
@@ -318,7 +295,6 @@ async fn handle_non_streaming_query(
             tenant_id.as_deref().unwrap_or(DEFAULT_TENANT),
         ])
         .observe(time);
-    record_query_metrics(time, query_range_seconds, QueryType::Sql);
     let response = QueryResponse {
         records,
         fields,

src/metrics/mod.rs

@@ -17,10 +17,7 @@
  */
 
 pub mod prom_utils;
-use std::{
-    path::Path,
-    sync::{Mutex, OnceLock},
-};
+use std::{path::Path, sync::OnceLock};
 
 use crate::{
     handlers::{TelemetryType, http::metrics_path},
@@ -42,145 +39,10 @@ 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] = &[
+pub 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,
 ];
 
-#[derive(Debug, PartialEq)]
-pub enum QueryType {
-    Sql,
-    Promql,
-}
-
-impl QueryType {
-    fn as_label(&self) -> &'static str {
-        match self {
-            Self::Sql => "SQL",
-            Self::Promql => "PromQL",
-        }
-    }
-}
-
-#[derive(Debug, PartialEq)]
-struct QueryMetricsSnapshot {
-    duration_seconds: f64,
-    time_range_seconds: f64,
-    query_type: QueryType,
-}
-
-#[derive(Default)]
-struct QueryMetricsAccumulator {
-    slowest_query: Mutex<Option<QueryMetricsSnapshot>>,
-}
-
-impl QueryMetricsAccumulator {
-    fn record(&self, duration_seconds: f64, time_range_seconds: f64, query_type: QueryType) {
-        let mut slowest_query = self
-            .slowest_query
-            .lock()
-            .expect("query metrics window lock should not be poisoned");
-
-        if slowest_query
-            .as_ref()
-            .is_none_or(|query| duration_seconds > query.duration_seconds)
-        {
-            *slowest_query = Some(QueryMetricsSnapshot {
-                duration_seconds,
-                time_range_seconds,
-                query_type,
-            });
-        }
-    }
-
-    fn take_and_reset(&self) -> Option<QueryMetricsSnapshot> {
-        let mut slowest_query = self
-            .slowest_query
-            .lock()
-            .expect("query metrics window lock should not be poisoned");
-        slowest_query.take()
-    }
-}
-
-static QUERY_METRICS_ACCUMULATOR: Lazy<QueryMetricsAccumulator> =
-    Lazy::new(QueryMetricsAccumulator::default);
-
-pub fn record_query_metrics(duration_seconds: f64, time_range_seconds: f64, query_type: QueryType) {
-    QUERY_METRICS_ACCUMULATOR.record(duration_seconds, time_range_seconds, query_type);
-}
-
-pub fn refresh_query_metrics() {
-    let snapshot = QUERY_METRICS_ACCUMULATOR.take_and_reset();
-    SLOWEST_QUERY_DURATION_SECONDS.reset();
-
-    if let Some(snapshot) = snapshot {
-        SLOWEST_QUERY_DURATION_SECONDS
-            .with_label_values(&[
-                &snapshot.time_range_seconds.to_string(),
-                snapshot.query_type.as_label(),
-            ])
-            .set(snapshot.duration_seconds);
-    }
-}
-
-#[cfg(test)]
-mod query_metrics_tests {
-    use prometheus::core::Collector;
-
-    use super::{
-        METRICS_NAMESPACE, METRICS_REGISTRY, QueryMetricsAccumulator, QueryMetricsSnapshot,
-        QueryType, SLOWEST_QUERY_DURATION_SECONDS, record_query_metrics, refresh_query_metrics,
-    };
-
-    #[test]
-    fn tracks_slowest_query_and_resets_after_snapshot() {
-        let accumulator = QueryMetricsAccumulator::default();
-
-        accumulator.record(20.0, 600.0, QueryType::Sql);
-        accumulator.record(70.0, 7_200.0, QueryType::Promql);
-        accumulator.record(65.0, 300.0, QueryType::Sql);
-
-        assert_eq!(
-            accumulator.take_and_reset(),
-            Some(QueryMetricsSnapshot {
-                duration_seconds: 70.0,
-                time_range_seconds: 7_200.0,
-                query_type: QueryType::Promql,
-            })
-        );
-        assert_eq!(accumulator.take_and_reset(), None);
-    }
-
-    #[test]
-    fn refreshes_gauges_and_clears_previous_window() {
-        record_query_metrics(20.0, 600.0, QueryType::Sql);
-        record_query_metrics(70.0, 7_200.0, QueryType::Promql);
-        refresh_query_metrics();
-
-        assert_eq!(
-            SLOWEST_QUERY_DURATION_SECONDS
-                .with_label_values(&["7200", "PromQL"])
-                .get(),
-            70.0
-        );
-        let metric_name = format!("{METRICS_NAMESPACE}_slowest_query_duration_seconds");
-        assert!(
-            METRICS_REGISTRY
-                .gather()
-                .iter()
-                .any(|family| family.get_name() == metric_name)
-        );
-
-        refresh_query_metrics();
-
-        assert!(
-            SLOWEST_QUERY_DURATION_SECONDS
-                .collect()
-                .iter()
-                .all(|family| family.get_metric().is_empty())
-        );
-    }
-}
-
 pub static METRICS_REGISTRY: Lazy<Registry> = Lazy::new(|| {
     let registry = Registry::new();
     custom_metrics(&registry);
@@ -492,18 +354,6 @@ pub static QUERY_EXECUTE_TIME: Lazy<HistogramVec> = Lazy::new(|| {
     .expect("metric can be created")
 });
 
-pub static SLOWEST_QUERY_DURATION_SECONDS: Lazy<GaugeVec> = Lazy::new(|| {
-    GaugeVec::new(
-        Opts::new(
-            "slowest_query_duration_seconds",
-            "Duration of the slowest query completed in the latest query metrics interval",
-        )
-        .namespace(METRICS_NAMESPACE),
-        &["time_range_seconds", "query_type"],
-    )
-    .expect("metric can be created")
-});
-
 pub static CONVERSION_TASK_DURATION: Lazy<HistogramVec> = Lazy::new(|| {
     HistogramVec::new(
         HistogramOpts::new(
@@ -1046,9 +896,6 @@ fn custom_metrics(registry: &Registry) {
     registry
         .register(Box::new(QUERY_EXECUTE_TIME.clone()))
         .expect("metric can be registered");
-    registry
-        .register(Box::new(SLOWEST_QUERY_DURATION_SECONDS.clone()))
-        .expect("metric can be registered");
     registry
         .register(Box::new(CONVERSION_TASK_DURATION.clone()))
         .expect("metric can be registered");