diff --git a/src/common/utils/http.rs b/src/common/utils/http.rs index 574a480ebf..56b9df7d6e 100644 --- a/src/common/utils/http.rs +++ b/src/common/utils/http.rs @@ -77,14 +77,10 @@ pub(crate) fn get_search_type_from_request( #[inline(always)] pub(crate) fn get_use_cache_from_request(query: &Query>) -> bool { - match query.get("use_cache") { - Some(s) => match s.to_lowercase().as_str() { - "true" => true, - "false" => false, - _ => true, - }, - None => true, - } + let Some(v) = query.get("use_cache") else { + return true; + }; + v.to_lowercase().as_str().parse::().unwrap_or(true) } #[inline(always)] diff --git a/src/handler/grpc/request/metrics/querier.rs b/src/handler/grpc/request/metrics/querier.rs index d8a5cede31..94fa944444 100644 --- a/src/handler/grpc/request/metrics/querier.rs +++ b/src/handler/grpc/request/metrics/querier.rs @@ -45,7 +45,7 @@ pub struct Querier; #[tonic::async_trait] impl Metrics for Querier { - #[tracing::instrument(name = "grpc:metrics:enter", skip_all, fields(org_id = req.get_ref().org_id))] + #[tracing::instrument(name = "grpc:metrics:query", skip_all, fields(org_id = req.get_ref().org_id))] async fn query( &self, req: Request, diff --git a/src/handler/grpc/request/search/mod.rs b/src/handler/grpc/request/search/mod.rs index 2ab7b8e6ed..f7dc48a035 100644 --- a/src/handler/grpc/request/search/mod.rs +++ b/src/handler/grpc/request/search/mod.rs @@ -115,7 +115,7 @@ impl Default for Searcher { #[tonic::async_trait] impl Search for Searcher { - #[tracing::instrument(name = "grpc:search:enter", skip_all, fields(org_id = req.get_ref().org_id))] + #[tracing::instrument(name = "grpc:search", skip_all, fields(org_id = req.get_ref().org_id))] async fn search( &self, req: Request, @@ -198,7 +198,7 @@ impl Search for Searcher { } } - #[tracing::instrument(name = "grpc:cluster_search:enter", skip_all, fields(org_id = req.get_ref().org_id))] + #[tracing::instrument(name = "grpc:cluster_search", skip_all, fields(org_id = req.get_ref().org_id))] async fn cluster_search( &self, req: Request, diff --git a/src/handler/http/request/search/mod.rs b/src/handler/http/request/search/mod.rs index 9a142d358f..e6b455d0ab 100644 --- a/src/handler/http/request/search/mod.rs +++ b/src/handler/http/request/search/mod.rs @@ -13,7 +13,7 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::{cmp, collections::HashMap, io::Error}; +use std::{collections::HashMap, io::Error}; use actix_web::{get, http::StatusCode, post, web, HttpRequest, HttpResponse}; use chrono::{Duration, Utc}; @@ -25,23 +25,15 @@ use config::{ usage::{RequestStats, UsageType}, }, metrics, - utils::{base64, hash::Sum64, json}, + utils::{base64, json}, DISTINCT_FIELDS, }; -use infra::{ - cache::{file_data::disk::QUERY_RESULT_CACHE, meta::ResultCacheMeta}, - errors, - schema::STREAM_SCHEMAS_LATEST, -}; +use infra::{errors, schema::STREAM_SCHEMAS_LATEST}; use tracing::{Instrument, Span}; use crate::{ common::{ - meta::{ - self, - http::HttpResponse as MetaHttpResponse, - search::{CachedQueryResponse, QueryDelta}, - }, + meta::{self, http::HttpResponse as MetaHttpResponse}, utils::{ functions, http::{ @@ -51,12 +43,8 @@ use crate::{ }, }, service::{ - search::{ - self as SearchService, - cache::{cacher::check_cache, result_utils::get_ts_value}, - sql::RE_ONLY_SELECT, - }, - usage::report_request_usage_stats, + search::{self as SearchService, sql::RE_ONLY_SELECT}, + usage::{http_report_metrics, report_request_usage_stats}, }, }; @@ -135,9 +123,6 @@ pub async fn search( in_req: HttpRequest, body: web::Bytes, ) -> Result { - let user_id_hdr = in_req.headers().get("user_id").unwrap(); - let user_id = user_id_hdr.to_str().unwrap().to_string(); - let started_at = Utc::now().timestamp_micros(); let start = std::time::Instant::now(); let org_id = org_id.into_inner(); let mut range_error = String::new(); @@ -148,6 +133,12 @@ pub async fn search( Span::none() }; let trace_id = get_or_create_trace_id(in_req.headers(), &http_span); + let user_id = in_req + .headers() + .get("user_id") + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(); let query = web::Query::>::from_query(in_req.query_string()).unwrap(); let stream_type = match get_stream_type_from_request(&query) { @@ -160,7 +151,7 @@ pub async fn search( Err(e) => return Ok(MetaHttpResponse::bad_request(e)), }; - let use_cache = get_use_cache_from_request(&query); + let use_cache = cfg.common.result_cache_enabled && get_use_cache_from_request(&query); // handle encoding for query and aggs let mut req: config::meta::search::Request = match json::from_slice(&body) { Ok(v) => v, @@ -170,9 +161,7 @@ pub async fn search( return Ok(MetaHttpResponse::bad_request(e)); } - let mut rpc_req: proto::cluster_rpc::SearchRequest = req.to_owned().into(); - rpc_req.org_id = org_id.to_string(); - rpc_req.stream_type = stream_type.to_string(); + // get stream name let parsed_sql = match config::meta::sql::Sql::new(&req.query.sql) { Ok(v) => v, Err(e) => { @@ -184,28 +173,23 @@ pub async fn search( ); } }; - let stream_name = &parsed_sql.source; - let r = STREAM_SCHEMAS_LATEST.read().await; - let stream_schema = r.get(format!("{}/{}/{}", org_id, stream_type, stream_name).as_str()); - if let Some(det) = stream_schema { - let local_schema = det.schema(); - if let Some(settings) = infra::schema::unwrap_stream_settings(local_schema) { - let max_query_range = settings.max_query_range; - if max_query_range > 0 - && (req.query.end_time - req.query.start_time) / (1000 * 1000 * 60 * 60) - > max_query_range - { - req.query.start_time = req.query.end_time - max_query_range * 1000 * 1000 * 60 * 60; - range_error = format!( - "Query duration is modified due to query range restriction of {} hours", - max_query_range - ); - } + // get stream settings + if let Some(settings) = infra::schema::get_settings(&org_id, stream_name, stream_type).await { + let max_query_range = settings.max_query_range; + if max_query_range > 0 + && (req.query.end_time - req.query.start_time) / (1000 * 1000 * 60 * 60) + > max_query_range + { + req.query.start_time = req.query.end_time - max_query_range * 1000 * 1000 * 60 * 60; + range_error = format!( + "Query duration is modified due to query range restriction of {} hours", + max_query_range + ); } } - drop(r); + // Check permissions on stream #[cfg(feature = "enterprise")] { @@ -239,268 +223,55 @@ pub async fn search( // Check permissions on stream ends } - // Result caching check start - let mut origin_sql = req.query.sql.clone(); - let is_aggregate = crate::service::search::cache::result_utils::is_aggregate_query(&origin_sql) - .unwrap_or_default(); - - let mut h = config::utils::hash::gxhash::new(); - let hashed_query = h.sum64(&origin_sql); - let mut file_path = format!( - "{}/{}/{}/{}", - org_id, stream_type, stream_name, hashed_query - ); - - let mut should_exec_query = true; - let mut ext_took_wait = 0; - - let mut c_resp: CachedQueryResponse = if use_cache && cfg.common.result_cache_enabled { - check_cache( - &rpc_req, - &mut req, - &mut origin_sql, - &parsed_sql, - &mut file_path, - is_aggregate, - &mut should_exec_query, - &trace_id, - ) - .await - } else { - CachedQueryResponse::default() - }; - - // No cache data present, add delta for full query - if !c_resp.has_cached_data { - c_resp.deltas.push(QueryDelta { - delta_start_time: req.query.start_time, - delta_end_time: req.query.end_time, - delta_removed_hits: false, - }) - } else { - log::info!( - "[trace_id {trace_id}] query deltas are: {:?}", - c_resp.deltas - ); - log::info!( - "[trace_id {trace_id}] Query original start time: {}, end time : {}", - req.query.start_time, - req.query.end_time - ); - } - // Result caching check ends - let mut results = Vec::new(); - let mut res = if should_exec_query { - let mut query_fn = req.query.query_fn.and_then(|v| base64::decode_url(&v).ok()); - if let Some(vrl_function) = &query_fn { - if !vrl_function.trim().ends_with('.') { - query_fn = Some(format!("{} \n .", vrl_function)); - } - } - req.query.query_fn = query_fn; - - for fn_name in functions::get_all_transform_keys(&org_id).await { - if req.query.sql.contains(&format!("{}(", fn_name)) { - req.query.uses_zo_fn = true; - break; - } - } - - metrics::QUERY_PENDING_NUMS - .with_label_values(&[&org_id]) - .inc(); - - // get a local search queue lock - #[cfg(not(feature = "enterprise"))] - let locker = SearchService::QUEUE_LOCKER.clone(); - #[cfg(not(feature = "enterprise"))] - let locker = locker.lock().await; - #[cfg(not(feature = "enterprise"))] - if !cfg.common.feature_query_queue_enabled { - drop(locker); - } - #[cfg(not(feature = "enterprise"))] - let took_wait = start.elapsed().as_millis() as usize; - #[cfg(feature = "enterprise")] - let took_wait = 0; - ext_took_wait = took_wait; - log::info!("http search API wait in queue took: {} ms", took_wait); - - metrics::QUERY_PENDING_NUMS - .with_label_values(&[&org_id]) - .dec(); - - let mut tasks = Vec::new(); - - for (i, delta) in c_resp.deltas.into_iter().enumerate() { - let http_span_local = http_span.clone(); - let mut req = req.clone(); - let org_id = org_id.clone(); - let trace_id = format!("{}-{:?}", trace_id.clone(), i); - let user_id = user_id.clone(); - - let task = tokio::task::spawn(async move { - let trace_id = trace_id.clone(); - req.query.start_time = delta.delta_start_time; - req.query.end_time = delta.delta_end_time; - let cfg = get_config(); - - if cfg.common.result_cache_enabled - && cfg.common.print_key_sql - && c_resp.has_cached_data - { - log::info!( - "[trace_id {trace_id}] Query new start time: {}, end time : {}", - req.query.start_time, - req.query.end_time - ); - } - - SearchService::search( - &trace_id, - &org_id, - stream_type, - Some(user_id.to_string()), - &req, - ) - .instrument(http_span_local) - .await - }) - .instrument(http_span.clone()); - tasks.push(task); - } - - for task in tasks { - match task.await { - Ok(res) => match res { - Ok(res) => results.push(res), - Err(err) => { - report_metrics(start, &org_id, stream_type, "", "500", "_search"); - log::error!("search error: {:?}", err); - return Ok(match err { - errors::Error::ErrorCode(code) => match code { - errors::ErrorCodes::SearchCancelQuery(_) => { - HttpResponse::TooManyRequests().json( - meta::http::HttpResponse::error_code_with_trace_id( - code, - Some(trace_id), - ), - ) - } - _ => HttpResponse::InternalServerError().json( - meta::http::HttpResponse::error_code_with_trace_id( - code, - Some(trace_id), - ), - ), - }, - _ => HttpResponse::InternalServerError().json( - meta::http::HttpResponse::error( - StatusCode::INTERNAL_SERVER_ERROR.into(), - err.to_string(), - ), - ), - }); - } - }, - Err(err) => { - report_metrics(start, &org_id, stream_type, "", "500", "_search"); - log::error!("search error: {:?}", err); - return Ok(HttpResponse::InternalServerError().json( - meta::http::HttpResponse::error( - StatusCode::INTERNAL_SERVER_ERROR.into(), - err.to_string(), - ), - )); - } - } - } - if c_resp.has_cached_data { - merge_response( - &mut c_resp.cached_response, - &results, - &c_resp.ts_column, - c_resp.limit, - c_resp.is_descending, - ); - c_resp.cached_response - } else { - results[0].clone() - } - } else { - c_resp.cached_response - }; - - // do search - let time = start.elapsed().as_secs_f64(); - report_metrics(start, &org_id, stream_type, "", "200", "_search"); - res.set_trace_id(trace_id.clone()); - res.set_local_took(start.elapsed().as_millis() as usize, ext_took_wait); - if !range_error.is_empty() { - res.is_partial = true; - res.function_error = if res.function_error.is_empty() { - range_error - } else { - format!("{} \n {}", range_error, res.function_error) - }; - res.new_start_time = Some(req.query.start_time); - res.new_end_time = Some(req.query.end_time); - } - - let req_stats = RequestStats { - records: res.hits.len() as i64, - response_time: time, - size: res.scan_size as f64, - request_body: Some(req.query.sql), - user_email: Some(user_id.to_string()), - min_ts: Some(req.query.start_time), - max_ts: Some(req.query.end_time), - cached_ratio: Some(res.cached_ratio), - search_type, - trace_id: Some(trace_id.clone()), - took_wait_in_queue: if res.took_detail.is_some() { - let resp_took = res.took_detail.as_ref().unwrap(); - // Consider only the cluster wait queue duration - Some(resp_took.cluster_wait_queue) - } else { - None - }, - ..Default::default() - }; - let num_fn = req.query.query_fn.is_some() as u16; - report_request_usage_stats( - req_stats, + // run search with cache + let res = SearchService::cache::search( + &trace_id, &org_id, - stream_name, - StreamType::Logs, - UsageType::Search, - num_fn, - started_at, + stream_type, + Some(user_id), + &req, + use_cache, + search_type, ) + .instrument(http_span) .await; - - // result cache save changes start - if cfg.common.result_cache_enabled - && should_exec_query - && c_resp.cache_query_response - && (!results.first().unwrap().hits.is_empty() || !results.last().unwrap().hits.is_empty()) - { - write_results( - &c_resp.ts_column, - req.query.start_time, - req.query.end_time, - &res, - file_path, - is_aggregate, - trace_id, - c_resp.is_descending, - ) - .await; + match res { + Ok(mut res) => { + if !range_error.is_empty() { + res.is_partial = true; + res.function_error = if res.function_error.is_empty() { + range_error + } else { + format!("{} \n {}", range_error, res.function_error) + }; + res.new_start_time = Some(req.query.start_time); + res.new_end_time = Some(req.query.end_time); + } + Ok(HttpResponse::Ok().json(res)) + } + Err(err) => { + http_report_metrics(start, &org_id, stream_type, "", "500", "_search"); + log::error!("[trace_id {trace_id}] search error: {}", err); + Ok(match err { + errors::Error::ErrorCode(code) => match code { + errors::ErrorCodes::SearchCancelQuery(_) => HttpResponse::TooManyRequests() + .json(meta::http::HttpResponse::error_code_with_trace_id( + code, + Some(trace_id), + )), + _ => HttpResponse::InternalServerError().json( + meta::http::HttpResponse::error_code_with_trace_id(code, Some(trace_id)), + ), + }, + _ => HttpResponse::InternalServerError().json(meta::http::HttpResponse::error( + StatusCode::INTERNAL_SERVER_ERROR.into(), + err.to_string(), + )), + }) + } } - // result cache save changes Ends - - Ok(HttpResponse::Ok().json(res)) } + /// SearchAround #[utoipa::path( context_path = "/api", @@ -565,6 +336,10 @@ pub async fn around( Span::none() }; let trace_id = get_or_create_trace_id(in_req.headers(), &http_span); + let user_id = in_req + .headers() + .get("user_id") + .map(|v| v.to_str().unwrap_or("").to_string()); let mut uses_fn = false; let query = web::Query::>::from_query(in_req.query_string()).unwrap(); @@ -685,13 +460,6 @@ pub async fn around( timeout, search_type: Some(SearchEventType::UI), }; - let user_id = in_req - .headers() - .get("user_id") - .unwrap() - .to_str() - .ok() - .map(|v| v.to_string()); let search_res = SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req) .instrument(http_span.clone()) .await; @@ -699,7 +467,7 @@ pub async fn around( let resp_forward = match search_res { Ok(res) => res, Err(err) => { - report_metrics(start, &org_id, stream_type, &stream_name, "500", "_around"); + http_report_metrics(start, &org_id, stream_type, &stream_name, "500", "_around"); log::error!("search around error: {:?}", err); return Ok(match err { errors::Error::ErrorCode(code) => match code { @@ -745,14 +513,14 @@ pub async fn around( timeout, search_type: Some(SearchEventType::UI), }; - let search_res = SearchService::search(&trace_id, &org_id, stream_type, user_id, &req) + let search_res = SearchService::search(&trace_id, &org_id, stream_type, user_id.clone(), &req) .instrument(http_span) .await; let resp_backward = match search_res { Ok(res) => res, Err(err) => { - report_metrics(start, &org_id, stream_type, &stream_name, "500", "_around"); + http_report_metrics(start, &org_id, stream_type, &stream_name, "500", "_around"); log::error!("search around error: {:?}", err); return Ok(match err { errors::Error::ErrorCode(code) => match code { @@ -791,18 +559,14 @@ pub async fn around( resp.cached_ratio = (resp_forward.cached_ratio + resp_backward.cached_ratio) / 2; let time = start.elapsed().as_secs_f64(); - report_metrics(start, &org_id, stream_type, &stream_name, "200", "_around"); + http_report_metrics(start, &org_id, stream_type, &stream_name, "200", "_around"); - let user_id = match in_req.headers().get("user_id") { - Some(v) => v.to_str().unwrap(), - None => "", - }; let req_stats = RequestStats { records: resp.hits.len() as i64, response_time: time, size: resp.scan_size as f64, request_body: Some(req.query.sql), - user_email: Some(user_id.to_string()), + user_email: user_id, min_ts: Some(around_start_time), max_ts: Some(around_end_time), cached_ratio: Some(resp.cached_ratio), @@ -890,10 +654,12 @@ pub async fn values( None => "".to_string(), Some(v) => base64::decode_url(v).unwrap_or("".to_string()), }; - let user_id = match in_req.headers().get("user_id") { - Some(v) => v.to_str().unwrap(), - None => "", - }; + let user_id = in_req + .headers() + .get("user_id") + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(); let http_span = if config::get_config().common.tracing_search_enabled { tracing::info_span!( "/api/{org_id}/{stream_name}/_values", @@ -921,7 +687,7 @@ pub async fn values( &fields[0], Some((column[0], column[1])), &query, - user_id, + &user_id, trace_id, http_span, ) @@ -936,7 +702,7 @@ pub async fn values( &fields[0], None, &query, - user_id, + &user_id, trace_id, http_span, ) @@ -951,7 +717,7 @@ pub async fn values( &fields[0], None, &query, - user_id, + &user_id, trace_id, http_span, ) @@ -964,7 +730,7 @@ pub async fn values( stream_type, &stream_name, &query, - user_id, + &user_id, trace_id, http_span, ) @@ -1164,7 +930,7 @@ async fn values_v1( let resp_search = match search_res { Ok(res) => res, Err(err) => { - report_metrics(start, org_id, stream_type, stream_name, "500", "_values/v1"); + http_report_metrics(start, org_id, stream_type, stream_name, "500", "_values/v1"); log::error!("search values error: {:?}", err); return Ok(match err { errors::Error::ErrorCode(code) => match code { @@ -1201,7 +967,7 @@ async fn values_v1( resp.cached_ratio = resp_search.cached_ratio; let time = start.elapsed().as_secs_f64(); - report_metrics(start, org_id, stream_type, stream_name, "200", "_values/v1"); + http_report_metrics(start, org_id, stream_type, stream_name, "200", "_values/v1"); let req_stats = RequestStats { records: resp.hits.len() as i64, @@ -1370,7 +1136,7 @@ async fn values_v2( let resp_search = match search_res { Ok(res) => res, Err(err) => { - report_metrics(start, org_id, stream_type, stream_name, "500", "_values/v2"); + http_report_metrics(start, org_id, stream_type, stream_name, "500", "_values/v2"); log::error!("search values error: {:?}", err); return Ok(match err { errors::Error::ErrorCode(code) => match code { @@ -1406,7 +1172,7 @@ async fn values_v2( resp.cached_ratio = resp_search.cached_ratio; let time = start.elapsed().as_secs_f64(); - report_metrics(start, org_id, stream_type, stream_name, "200", "_values/v2"); + http_report_metrics(start, org_id, stream_type, stream_name, "200", "_values/v2"); let req_stats = RequestStats { records: resp.hits.len() as i64, @@ -1511,11 +1277,11 @@ pub async fn search_partition( // do search match search_res { Ok(res) => { - report_metrics(start, &org_id, stream_type, "", "200", "_search_partition"); + http_report_metrics(start, &org_id, stream_type, "", "200", "_search_partition"); Ok(HttpResponse::Ok().json(res)) } Err(err) => { - report_metrics(start, &org_id, stream_type, "", "500", "_search_partition"); + http_report_metrics(start, &org_id, stream_type, "", "500", "_search_partition"); log::error!("search error: {:?}", err); Ok(match err { errors::Error::ErrorCode(code) => HttpResponse::InternalServerError().json( @@ -1529,185 +1295,3 @@ pub async fn search_partition( } } } - -// based on _timestamp of first record in config::meta::search::Response either add it in start -// or end to cache response -fn merge_response( - cache_response: &mut config::meta::search::Response, - search_response: &Vec, - ts_column: &str, - limit: i64, - is_descending: bool, -) { - if cache_response.hits.is_empty() && search_response.is_empty() { - return; - } - - if cache_response.hits.is_empty() - && search_response.is_empty() - && search_response.first().unwrap().hits.is_empty() - && search_response.last().unwrap().hits.is_empty() - { - for res in search_response { - cache_response.total += res.total; - cache_response.scan_size += res.scan_size; - cache_response.took += res.took; - } - return; - } - let cache_hits_len = cache_response.hits.len(); - - let mut files_cache_ratio = 0; - let mut result_cache_len = 0; - let cache_ts = if cache_response.hits.is_empty() { - get_ts_value( - ts_column, - search_response.first().unwrap().hits.first().unwrap(), - ) - } else { - get_ts_value(ts_column, cache_response.hits.first().unwrap()) - }; - - for res in search_response { - cache_response.total += res.total; - cache_response.scan_size += res.scan_size; - cache_response.took += res.took; - files_cache_ratio += res.cached_ratio; - - result_cache_len += res.total; - - if res.hits.is_empty() { - continue; - } - let search_ts = get_ts_value(ts_column, res.hits.first().unwrap()); - if search_ts < cache_ts && is_descending { - cache_response.hits.extend(res.hits.clone()); - } else { - cache_response.hits = res - .hits - .iter() - .chain(cache_response.hits.iter()) - .cloned() - .collect(); - } - } - if is_descending { - cache_response - .hits - .sort_by_key(|b| std::cmp::Reverse(get_ts_value(ts_column, b))); - } else { - cache_response - .hits - .sort_by_key(|a| get_ts_value(ts_column, a)); - } - - if cache_response.hits.len() > limit as usize { - cache_response.hits.truncate(limit as usize); - } - cache_response.cached_ratio = files_cache_ratio / search_response.len(); - log::info!( - "cache_response.hits.len: {}, Result cache len: {}", - cache_hits_len, - result_cache_len, - ); - cache_response.result_cache_ratio = - (cache_hits_len as f64 * 100_f64 / (result_cache_len + cache_hits_len) as f64) as usize; -} - -#[allow(clippy::too_many_arguments)] -async fn write_results( - ts_column: &str, - req_query_start_time: i64, - req_query_end_time: i64, - res: &config::meta::search::Response, - file_path: String, - is_aggregate: bool, - trace_id: String, - is_descending: bool, -) { - let last_rec_ts = get_ts_value(ts_column, res.hits.last().unwrap()); - let first_rec_ts = get_ts_value(ts_column, res.hits.first().unwrap()); - - if (last_rec_ts - first_rec_ts).abs() - < get_config().common.result_cache_discard_duration * 1000 * 1000 - { - return; - } - - let largest_ts = cmp::max(first_rec_ts, last_rec_ts); - - let cache_end_time = if largest_ts > 0 && largest_ts < req_query_end_time { - last_rec_ts - } else { - req_query_end_time - }; - let file_name = format!( - "{}_{}_{}_{}.json", - req_query_start_time, - cache_end_time, - if is_aggregate { 1 } else { 0 }, - if is_descending { 1 } else { 0 } - ); - - let res_cache = json::to_string(&res).unwrap(); - let query_key = file_path.replace('/', "_"); - tokio::spawn(async move { - let file_path_local = file_path.clone(); - - match SearchService::cache::cacher::cache_results_to_disk( - &trace_id, - &file_path_local, - &file_name, - res_cache, - ) - .await - { - Ok(_) => { - let mut w = QUERY_RESULT_CACHE.write().await; - w.entry(query_key) - .or_insert_with(Vec::new) - .push(ResultCacheMeta { - start_time: req_query_start_time, - end_time: cache_end_time, - is_aggregate, - is_descending, - }); - drop(w); - } - Err(e) => { - log::error!("Cache results to disk failed: {:?}", e); - } - } - }); -} - -#[inline] -fn report_metrics( - start: std::time::Instant, - org_id: &str, - stream_type: StreamType, - stream_name: &str, - code: &str, - uri: &str, -) { - let time = start.elapsed().as_secs_f64(); - let uri = format!("/api/org/{}", uri); - metrics::HTTP_RESPONSE_TIME - .with_label_values(&[ - &uri, - code, - org_id, - stream_name, - stream_type.to_string().as_str(), - ]) - .observe(time); - metrics::HTTP_INCOMING_REQUESTS - .with_label_values(&[ - &uri, - code, - org_id, - stream_name, - stream_type.to_string().as_str(), - ]) - .inc(); -} diff --git a/src/infra/src/schema/mod.rs b/src/infra/src/schema/mod.rs index f2824a42f4..8144eaeaf8 100644 --- a/src/infra/src/schema/mod.rs +++ b/src/infra/src/schema/mod.rs @@ -113,6 +113,7 @@ pub async fn get_from_db( }) } +#[tracing::instrument(name = "infra:schema:get_versions", skip_all)] pub async fn get_versions( org_id: &str, stream_name: &str, diff --git a/src/service/search/cache/cacher.rs b/src/service/search/cache/cacher.rs index 5752ed3740..eada48bd95 100644 --- a/src/service/search/cache/cacher.rs +++ b/src/service/search/cache/cacher.rs @@ -40,6 +40,7 @@ use crate::{ )] #[allow(clippy::too_many_arguments)] pub async fn check_cache( + trace_id: &str, rpc_req: &proto::cluster_rpc::SearchRequest, req: &mut config::meta::search::Request, origin_sql: &mut String, @@ -47,7 +48,6 @@ pub async fn check_cache( file_path: &mut String, is_aggregate: bool, should_exec_query: &mut bool, - trace_id: &str, ) -> CachedQueryResponse { let start = std::time::Instant::now(); let cfg = get_config(); diff --git a/src/service/search/cache/mod.rs b/src/service/search/cache/mod.rs index 80e34971e3..a18fdb13b1 100644 --- a/src/service/search/cache/mod.rs +++ b/src/service/search/cache/mod.rs @@ -13,5 +13,429 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . +use chrono::Utc; +use config::{ + get_config, + meta::{ + search::{self, SearchEventType}, + stream::StreamType, + usage::{RequestStats, UsageType}, + }, + metrics, + utils::{base64, hash::Sum64, json}, +}; +use infra::{ + cache::{file_data::disk::QUERY_RESULT_CACHE, meta::ResultCacheMeta}, + errors::Error, +}; +use result_utils::get_ts_value; +use tracing::Instrument; + +use crate::{ + common::{ + meta::search::{CachedQueryResponse, QueryDelta}, + utils::functions, + }, + service::{ + search::{ + self as SearchService, + cache::{cacher::check_cache, result_utils::is_aggregate_query}, + }, + usage::{http_report_metrics, report_request_usage_stats}, + }, +}; + pub mod cacher; pub mod result_utils; + +#[tracing::instrument(name = "service:search:cacher:search", skip_all)] +pub async fn search( + trace_id: &str, + org_id: &str, + stream_type: StreamType, + user_id: Option, + in_req: &search::Request, + use_cache: bool, + search_type: Option, +) -> Result { + let started_at = Utc::now().timestamp_micros(); + let start = std::time::Instant::now(); + let cfg = get_config(); + + // Result caching check start + let mut origin_sql = in_req.query.sql.clone(); + let is_aggregate = is_aggregate_query(&origin_sql).unwrap_or_default(); + let parsed_sql = match config::meta::sql::Sql::new(&origin_sql) { + Ok(v) => v, + Err(e) => { + return Err(Error::Message(e.to_string())); + } + }; + let stream_name = &parsed_sql.source; + + let mut h = config::utils::hash::gxhash::new(); + let hashed_query = h.sum64(&origin_sql); + + let mut should_exec_query = true; + let mut ext_took_wait = 0; + + let mut req = in_req.clone(); + let mut rpc_req: proto::cluster_rpc::SearchRequest = req.to_owned().into(); + rpc_req.org_id = org_id.to_string(); + rpc_req.stream_type = stream_type.to_string(); + + let mut file_path = format!( + "{}/{}/{}/{}", + org_id, stream_type, stream_name, hashed_query + ); + let mut c_resp: CachedQueryResponse = if use_cache { + check_cache( + trace_id, + &rpc_req, + &mut req, + &mut origin_sql, + &parsed_sql, + &mut file_path, + is_aggregate, + &mut should_exec_query, + ) + .await + } else { + CachedQueryResponse::default() + }; + + // No cache data present, add delta for full query + if !c_resp.has_cached_data { + c_resp.deltas.push(QueryDelta { + delta_start_time: req.query.start_time, + delta_end_time: req.query.end_time, + delta_removed_hits: false, + }) + } else if use_cache { + log::info!( + "[trace_id {trace_id}] Query deltas are: {:?}", + c_resp.deltas + ); + log::info!( + "[trace_id {trace_id}] Query original start time: {}, end time : {}", + req.query.start_time, + req.query.end_time + ); + } + + // Result caching check ends, start search + let mut results = Vec::new(); + let mut res = if !should_exec_query { + c_resp.cached_response + } else { + let mut query_fn = req.query.query_fn.and_then(|v| base64::decode_url(&v).ok()); + if let Some(vrl_function) = &query_fn { + if !vrl_function.trim().ends_with('.') { + query_fn = Some(format!("{} \n .", vrl_function)); + } + } + req.query.query_fn = query_fn; + + for fn_name in functions::get_all_transform_keys(org_id).await { + if req.query.sql.contains(&format!("{}(", fn_name)) { + req.query.uses_zo_fn = true; + break; + } + } + + metrics::QUERY_PENDING_NUMS + .with_label_values(&[org_id]) + .inc(); + + // get a local search queue lock + #[cfg(not(feature = "enterprise"))] + let locker = SearchService::QUEUE_LOCKER.clone(); + #[cfg(not(feature = "enterprise"))] + let locker = locker.lock().await; + #[cfg(not(feature = "enterprise"))] + if !cfg.common.feature_query_queue_enabled { + drop(locker); + } + #[cfg(not(feature = "enterprise"))] + let took_wait = start.elapsed().as_millis() as usize; + #[cfg(feature = "enterprise")] + let took_wait = 0; + ext_took_wait = took_wait; + log::info!( + "[trace_id {trace_id}] http search API wait in queue took: {} ms", + took_wait + ); + + metrics::QUERY_PENDING_NUMS + .with_label_values(&[org_id]) + .dec(); + + let mut tasks = Vec::new(); + + let delta_len = c_resp.deltas.len(); + for (i, delta) in c_resp.deltas.into_iter().enumerate() { + let mut req = req.clone(); + let org_id = org_id.to_string(); + let trace_id = if delta_len <= 1 { + trace_id.to_string() + } else { + format!("{}-{}", trace_id, i) + }; + let user_id = user_id.clone(); + + let enter_span = tracing::span::Span::current(); + let task = tokio::task::spawn(async move { + let trace_id = trace_id.clone(); + req.query.start_time = delta.delta_start_time; + req.query.end_time = delta.delta_end_time; + + let cfg = get_config(); + if cfg.common.result_cache_enabled + && cfg.common.print_key_sql + && c_resp.has_cached_data + { + log::info!( + "[trace_id {trace_id}] Query new start time: {}, end time : {}", + req.query.start_time, + req.query.end_time + ); + } + + SearchService::search(&trace_id, &org_id, stream_type, user_id, &req) + .instrument(enter_span) + .await + }); + tasks.push(task); + } + + for task in tasks { + results.push(task.await.map_err(|e| Error::Message(e.to_string()))??); + } + if c_resp.has_cached_data { + merge_response( + trace_id, + &mut c_resp.cached_response, + &results, + &c_resp.ts_column, + c_resp.limit, + c_resp.is_descending, + ); + c_resp.cached_response + } else { + results[0].clone() + } + }; + + // do search + let time = start.elapsed().as_secs_f64(); + http_report_metrics(start, org_id, stream_type, "", "200", "_search"); + res.set_trace_id(trace_id.to_string()); + res.set_local_took(start.elapsed().as_millis() as usize, ext_took_wait); + + let req_stats = RequestStats { + records: res.hits.len() as i64, + response_time: time, + size: res.scan_size as f64, + request_body: Some(req.query.sql), + user_email: user_id, + min_ts: Some(req.query.start_time), + max_ts: Some(req.query.end_time), + cached_ratio: Some(res.cached_ratio), + search_type, + trace_id: Some(trace_id.to_string()), + took_wait_in_queue: if res.took_detail.is_some() { + let resp_took = res.took_detail.as_ref().unwrap(); + // Consider only the cluster wait queue duration + Some(resp_took.cluster_wait_queue) + } else { + None + }, + ..Default::default() + }; + let num_fn = req.query.query_fn.is_some() as u16; + report_request_usage_stats( + req_stats, + org_id, + stream_name, + StreamType::Logs, + UsageType::Search, + num_fn, + started_at, + ) + .await; + + // result cache save changes start + if cfg.common.result_cache_enabled + && should_exec_query + && c_resp.cache_query_response + && (!results.first().unwrap().hits.is_empty() || !results.last().unwrap().hits.is_empty()) + { + write_results( + trace_id, + &c_resp.ts_column, + req.query.start_time, + req.query.end_time, + &res, + file_path, + is_aggregate, + c_resp.is_descending, + ) + .await; + } + // result cache save changes Ends + + Ok(res) +} + +// based on _timestamp of first record in config::meta::search::Response either add it in start +// or end to cache response +fn merge_response( + trace_id: &str, + cache_response: &mut config::meta::search::Response, + search_response: &Vec, + ts_column: &str, + limit: i64, + is_descending: bool, +) { + if cache_response.hits.is_empty() && search_response.is_empty() { + return; + } + + if cache_response.hits.is_empty() + && search_response.is_empty() + && search_response.first().unwrap().hits.is_empty() + && search_response.last().unwrap().hits.is_empty() + { + for res in search_response { + cache_response.total += res.total; + cache_response.scan_size += res.scan_size; + cache_response.took += res.took; + } + return; + } + let cache_hits_len = cache_response.hits.len(); + + let mut files_cache_ratio = 0; + let mut result_cache_len = 0; + let cache_ts = if cache_response.hits.is_empty() { + get_ts_value( + ts_column, + search_response.first().unwrap().hits.first().unwrap(), + ) + } else { + get_ts_value(ts_column, cache_response.hits.first().unwrap()) + }; + + for res in search_response { + cache_response.total += res.total; + cache_response.scan_size += res.scan_size; + cache_response.took += res.took; + files_cache_ratio += res.cached_ratio; + + result_cache_len += res.total; + + if res.hits.is_empty() { + continue; + } + let search_ts = get_ts_value(ts_column, res.hits.first().unwrap()); + if search_ts < cache_ts && is_descending { + cache_response.hits.extend(res.hits.clone()); + } else { + cache_response.hits = res + .hits + .iter() + .chain(cache_response.hits.iter()) + .cloned() + .collect(); + } + } + if is_descending { + cache_response + .hits + .sort_by_key(|b| std::cmp::Reverse(get_ts_value(ts_column, b))); + } else { + cache_response + .hits + .sort_by_key(|a| get_ts_value(ts_column, a)); + } + + if cache_response.hits.len() > limit as usize { + cache_response.hits.truncate(limit as usize); + } + cache_response.cached_ratio = files_cache_ratio / search_response.len(); + log::info!( + "[trace_id {trace_id}] cache_response.hits.len: {}, Result cache len: {}", + cache_hits_len, + result_cache_len, + ); + cache_response.result_cache_ratio = + (cache_hits_len as f64 * 100_f64 / (result_cache_len + cache_hits_len) as f64) as usize; +} + +#[allow(clippy::too_many_arguments)] +async fn write_results( + trace_id: &str, + ts_column: &str, + req_query_start_time: i64, + req_query_end_time: i64, + res: &config::meta::search::Response, + file_path: String, + is_aggregate: bool, + is_descending: bool, +) { + let last_rec_ts = get_ts_value(ts_column, res.hits.last().unwrap()); + let first_rec_ts = get_ts_value(ts_column, res.hits.first().unwrap()); + + if (last_rec_ts - first_rec_ts).abs() + < get_config().common.result_cache_discard_duration * 1000 * 1000 + { + return; + } + + let largest_ts = std::cmp::max(first_rec_ts, last_rec_ts); + + let cache_end_time = if largest_ts > 0 && largest_ts < req_query_end_time { + last_rec_ts + } else { + req_query_end_time + }; + let file_name = format!( + "{}_{}_{}_{}.json", + req_query_start_time, + cache_end_time, + if is_aggregate { 1 } else { 0 }, + if is_descending { 1 } else { 0 } + ); + + let res_cache = json::to_string(&res).unwrap(); + let query_key = file_path.replace('/', "_"); + let trace_id = trace_id.to_string(); + tokio::spawn(async move { + let file_path_local = file_path.clone(); + + match SearchService::cache::cacher::cache_results_to_disk( + &trace_id, + &file_path_local, + &file_name, + res_cache, + ) + .await + { + Ok(_) => { + let mut w = QUERY_RESULT_CACHE.write().await; + w.entry(query_key) + .or_insert_with(Vec::new) + .push(ResultCacheMeta { + start_time: req_query_start_time, + end_time: cache_end_time, + is_aggregate, + is_descending, + }); + drop(w); + } + Err(e) => { + log::error!("Cache results to disk failed: {:?}", e); + } + } + }); +} diff --git a/src/service/search/cache/result_utils.rs b/src/service/search/cache/result_utils.rs index 5efcdf05d1..7bcabc4523 100644 --- a/src/service/search/cache/result_utils.rs +++ b/src/service/search/cache/result_utils.rs @@ -13,11 +13,7 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use std::io::{self, Read, Write}; - -use base64::Engine; use config::utils::{json, time::parse_str_to_timestamp_micros_as_option}; -use flate2::{read::ZlibDecoder, write::ZlibEncoder, Compression}; use sqlparser::{ ast::{Expr, Function, Query, SelectItem, SetExpr, Statement}, dialect::GenericDialect, @@ -72,27 +68,6 @@ const AGGREGATE_UDF_LIST: [&str; 7] = [ "approx_percentile_cont", ]; -pub fn encode_sql_to_foldername(sql_query: &str) -> io::Result { - let mut encoder = ZlibEncoder::new(Vec::new(), Compression::best()); - encoder.write_all(sql_query.as_bytes())?; - let compressed_bytes = encoder.finish()?; - - Ok(base64::engine::general_purpose::STANDARD.encode(compressed_bytes)) -} - -pub fn decode_foldername_to_sql(folder_name: &str) -> io::Result { - let compressed_bytes = base64::engine::general_purpose::STANDARD - .decode(folder_name.as_bytes()) - .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; - - let mut decoder = ZlibDecoder::new(&compressed_bytes[..]); - let mut decompressed_bytes = Vec::new(); - - decoder.read_to_end(&mut decompressed_bytes)?; - - String::from_utf8(decompressed_bytes).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e)) -} - pub fn get_ts_value(ts_column: &str, record: &json::Value) -> i64 { match record.get(ts_column).unwrap() { serde_json::Value::String(ts) => { diff --git a/src/service/search/grpc/mod.rs b/src/service/search/grpc/mod.rs index 95806e03b4..6cc520609f 100644 --- a/src/service/search/grpc/mod.rs +++ b/src/service/search/grpc/mod.rs @@ -29,7 +29,7 @@ use futures::future::try_join_all; use hashbrown::HashMap; use infra::errors::{Error, ErrorCodes}; use proto::cluster_rpc; -use tracing::{info_span, Instrument}; +use tracing::Instrument; use super::{datafusion, sql::Sql}; use crate::service::db; @@ -77,43 +77,31 @@ pub async fn search( let work_group1 = work_group.clone(); let trace_id1 = trace_id.clone(); let sql1 = sql.clone(); - let wal_parquet_span = info_span!( - "service:search:grpc:in_wal_parquet", - org_id = sql.org_id, - stream_name = sql.stream_name, - stream_type = stream_type.to_string(), - ); - let task1 = tokio::task::spawn( - async move { - if cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) && !skip_wal { - wal::search_parquet(&trace_id1, sql1, stream_type, &work_group1, timeout).await - } else { - Ok((HashMap::new(), ScanStats::default())) - } + let wal_parquet_span = tracing::span::Span::current(); + let task1 = tokio::task::spawn(async move { + if cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) && !skip_wal { + wal::search_parquet(&trace_id1, sql1, stream_type, &work_group1, timeout) + .instrument(wal_parquet_span) + .await + } else { + Ok((HashMap::new(), ScanStats::default())) } - .instrument(wal_parquet_span), - ); + }); // search in WAL memory let work_group2 = work_group.clone(); let trace_id2 = trace_id.clone(); let sql2 = sql.clone(); - let wal_mem_span = info_span!( - "service:search:grpc:in_wal_memory", - org_id = sql.org_id, - stream_name = sql.stream_name, - stream_type = stream_type.to_string(), - ); - let task2 = tokio::task::spawn( - async move { - if cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) && !skip_wal { - wal::search_memtable(&trace_id2, sql2, stream_type, &work_group2, timeout).await - } else { - Ok((HashMap::new(), ScanStats::default())) - } + let wal_mem_span = tracing::span::Span::current(); + let task2 = tokio::task::spawn(async move { + if cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) && !skip_wal { + wal::search_memtable(&trace_id2, sql2, stream_type, &work_group2, timeout) + .instrument(wal_mem_span) + .await + } else { + Ok((HashMap::new(), ScanStats::default())) } - .instrument(wal_mem_span), - ); + }); // search in object storage let req_stype = req.stype; @@ -121,30 +109,23 @@ pub async fn search( let trace_id3 = trace_id.clone(); let sql3 = sql.clone(); let file_list: Vec = req.file_list.iter().map(FileKey::from).collect(); - let storage_span = info_span!( - "service:search:grpc:in_storage", - org_id = sql.org_id, - stream_name = sql.stream_name, - stream_type = stream_type.to_string(), - ); - let task3 = tokio::task::spawn( - async move { - if req_stype == cluster_rpc::SearchType::WalOnly as i32 { - Ok((HashMap::new(), ScanStats::default())) - } else { - storage::search( - &trace_id3, - sql3, - &file_list, - stream_type, - &work_group3, - timeout, - ) - .await - } + let storage_span = tracing::span::Span::current(); + let task3 = tokio::task::spawn(async move { + if req_stype == cluster_rpc::SearchType::WalOnly as i32 { + Ok((HashMap::new(), ScanStats::default())) + } else { + storage::search( + &trace_id3, + sql3, + &file_list, + stream_type, + &work_group3, + timeout, + ) + .instrument(storage_span) + .await } - .instrument(storage_span), - ); + }); // merge result let mut results = HashMap::new(); diff --git a/src/service/search/grpc/storage.rs b/src/service/search/grpc/storage.rs index 96673bcef6..4cc4ec7a8b 100644 --- a/src/service/search/grpc/storage.rs +++ b/src/service/search/grpc/storage.rs @@ -47,7 +47,7 @@ use crate::service::{ type CachedFiles = (usize, usize); /// search in remote object storage -#[tracing::instrument(name = "service:search:grpc:storage:enter", skip_all, fields(org_id = sql.org_id, stream_name = sql.stream_name))] +#[tracing::instrument(name = "service:search:grpc:storage", skip_all, fields(org_id = sql.org_id, stream_name = sql.stream_name))] pub async fn search( trace_id: &str, sql: Arc, @@ -56,6 +56,7 @@ pub async fn search( work_group: &str, timeout: u64, ) -> super::SearchResult { + let enter_span = tracing::span::Span::current(); log::info!("[trace_id {trace_id}] search->storage: enter"); let schema_latest = infra::schema::get(&sql.org_id, &sql.stream_name, stream_type) .await @@ -67,6 +68,7 @@ pub async fn search( stream_type, sql.meta.time_range, ) + .instrument(enter_span.clone()) .await { Ok(versions) => versions, @@ -104,6 +106,7 @@ pub async fn search( partition_time_level, &stream_settings.partition_keys, ) + .instrument(enter_span.clone()) .await? } false => file_list.to_vec(), @@ -181,7 +184,9 @@ pub async fn search( // load files to local cache let (cache_type, deleted_files, (mem_cached_files, disk_cached_files)) = - cache_parquet_files(trace_id, &files, &scan_stats).await?; + cache_parquet_files(trace_id, &files, &scan_stats) + .instrument(enter_span.clone()) + .await?; if !deleted_files.is_empty() { // remove deleted files from files_group for (_, g_files) in files_group.iter_mut() { diff --git a/src/service/search/grpc/wal.rs b/src/service/search/grpc/wal.rs index 25d312b2de..9c44e7371c 100644 --- a/src/service/search/grpc/wal.rs +++ b/src/service/search/grpc/wal.rs @@ -52,7 +52,7 @@ use crate::{ }; /// search in local WAL, which haven't been sync to object storage -#[tracing::instrument(name = "service:search:wal:parquet::enter", skip_all, fields(org_id = sql.org_id, stream_name = sql.stream_name))] +#[tracing::instrument(name = "service:search:wal:parquet", skip_all, fields(org_id = sql.org_id, stream_name = sql.stream_name))] pub async fn search_parquet( trace_id: &str, sql: Arc, @@ -386,7 +386,7 @@ pub async fn search_parquet( } /// search in local WAL, which haven't been sync to object storage -#[tracing::instrument(name = "service:search:wal:mem:enter", skip_all, fields(org_id = sql.org_id, stream_name = sql.stream_name))] +#[tracing::instrument(name = "service:search:wal:memtable", skip_all, fields(org_id = sql.org_id, stream_name = sql.stream_name))] pub async fn search_memtable( trace_id: &str, sql: Arc, diff --git a/src/service/search/mod.rs b/src/service/search/mod.rs index b198fd5d34..099a4c028c 100644 --- a/src/service/search/mod.rs +++ b/src/service/search/mod.rs @@ -52,7 +52,7 @@ use crate::{ service::format_partition_key, }; -pub mod cache; +pub(crate) mod cache; pub(crate) mod cluster; pub(crate) mod datafusion; pub(crate) mod grpc; @@ -212,7 +212,7 @@ pub async fn search( } } -#[tracing::instrument(name = "service:search_partition:enter", skip(req))] +#[tracing::instrument(name = "service:search_partition", skip(req))] pub async fn search_partition( trace_id: &str, org_id: &str, @@ -656,7 +656,7 @@ pub fn server_internal_error(error: impl ToString) -> Error { Error::ErrorCode(ErrorCodes::ServerInternalError(error.to_string())) } -#[tracing::instrument(name = "service:search_partition_multi:enter", skip(req))] +#[tracing::instrument(name = "service:search_partition_multi", skip(req))] pub async fn search_partition_multi( trace_id: &str, org_id: &str, diff --git a/src/service/search/sql.rs b/src/service/search/sql.rs index 8bdfed58fa..403a2086f7 100644 --- a/src/service/search/sql.rs +++ b/src/service/search/sql.rs @@ -912,54 +912,23 @@ pub fn generate_histogram_interval(time_range: Option<(i64, i64)>, num: u16) -> } let intervals = [ - ( - Duration::try_hours(24 * 30) - .unwrap() - .num_microseconds() - .unwrap(), - "1 day", - ), - ( - Duration::try_hours(24 * 7) - .unwrap() - .num_microseconds() - .unwrap(), - "1 hour", - ), - ( - Duration::try_hours(24).unwrap().num_microseconds().unwrap(), - "30 minute", - ), - ( - Duration::try_hours(6).unwrap().num_microseconds().unwrap(), - "5 minute", - ), - ( - Duration::try_hours(2).unwrap().num_microseconds().unwrap(), - "1 minute", - ), - ( - Duration::try_hours(1).unwrap().num_microseconds().unwrap(), - "30 second", - ), - ( - Duration::try_minutes(30) - .unwrap() - .num_microseconds() - .unwrap(), - "15 second", - ), - ( - Duration::try_minutes(15) - .unwrap() - .num_microseconds() - .unwrap(), - "10 second", - ), + (Duration::try_hours(24 * 60), "1 day"), + (Duration::try_hours(24 * 30), "12 hour"), + (Duration::try_hours(24 * 28), "6 hour"), + (Duration::try_hours(24 * 21), "3 hour"), + (Duration::try_hours(24 * 14), "2 hour"), + (Duration::try_hours(24 * 7), "1 hour"), + (Duration::try_hours(24), "30 minute"), + (Duration::try_hours(6), "5 minute"), + (Duration::try_hours(2), "1 minute"), + (Duration::try_hours(1), "30 second"), + (Duration::try_minutes(30), "15 second"), + (Duration::try_minutes(15), "10 second"), ]; - for interval in intervals.iter() { - if (time_range.1 - time_range.0) >= interval.0 { - return interval.1.to_string(); + for (time, interval) in intervals.iter() { + let time = time.unwrap().num_microseconds().unwrap(); + if (time_range.1 - time_range.0) >= time { + return interval.to_string(); } } "10 second".to_string() diff --git a/src/service/usage/mod.rs b/src/service/usage/mod.rs index a29b3b1b8e..9e9292791d 100644 --- a/src/service/usage/mod.rs +++ b/src/service/usage/mod.rs @@ -452,3 +452,34 @@ async fn publish_audit( ) -> Result { ingestion_service::ingest(&get_config().common.usage_org, req).await } + +#[inline] +pub fn http_report_metrics( + start: std::time::Instant, + org_id: &str, + stream_type: StreamType, + stream_name: &str, + code: &str, + uri: &str, +) { + let time = start.elapsed().as_secs_f64(); + let uri = format!("/api/org/{}", uri); + metrics::HTTP_RESPONSE_TIME + .with_label_values(&[ + &uri, + code, + org_id, + stream_name, + stream_type.to_string().as_str(), + ]) + .observe(time); + metrics::HTTP_INCOMING_REQUESTS + .with_label_values(&[ + &uri, + code, + org_id, + stream_name, + stream_type.to_string().as_str(), + ]) + .inc(); +}