diff --git a/src/common/infra/cluster/mod.rs b/src/common/infra/cluster/mod.rs index b1860e1788..39f680f800 100644 --- a/src/common/infra/cluster/mod.rs +++ b/src/common/infra/cluster/mod.rs @@ -590,6 +590,9 @@ async fn set_node_status_metrics(node: &Node) { fn update_node_status_metrics() -> NodeMetrics { let node_status = get_node_metrics(); + config::metrics::NODE_UP + .with_label_values(&[config::VERSION]) + .set(1); config::metrics::NODE_CPU_TOTAL .with_label_values(&[]) .set(node_status.cpu_total as i64); diff --git a/src/common/utils/auth_tests.rs b/src/common/utils/auth_tests.rs index 52ccdbe10b..f69a313dbb 100644 --- a/src/common/utils/auth_tests.rs +++ b/src/common/utils/auth_tests.rs @@ -3014,6 +3014,7 @@ mod tests { app_name: String::default(), local_mode: bool::default(), local_mode_storage: String::default(), + is_local_storage: true, cluster_coordinator: String::default(), queue_store: String::default(), meta_store: String::default(), @@ -3032,6 +3033,8 @@ mod tests { data_db_dir: String::default(), data_cache_dir: String::default(), column_all: String::default(), + parquet_compression: String::default(), + timestamp_compression_disabled: bool::default(), feature_per_thread_lock: bool::default(), feature_fulltext_extra_fields: String::default(), feature_secondary_index_extra_fields: String::default(), @@ -3050,7 +3053,6 @@ mod tests { feature_query_skip_wal: bool::default(), wal_write_queue_enabled: bool::default(), wal_write_queue_full_reject: bool::default(), - websocket_close_frame_delay: u64::default(), ui_enabled: bool::default(), ui_sql_base64_enabled: bool::default(), metrics_dedup_enabled: bool::default(), @@ -3059,8 +3061,8 @@ mod tests { bloom_filter_default_fields: String::default(), bloom_filter_ndv_ratio: u64::default(), wal_fsync_disabled: bool::default(), - wal_write_queue_full_reject: bool::default(), - wal_write_queue_enabled: bool::default(), + search_around_default_fields: String::default(), + additional_reporting_orgs: String::default(), tracing_enabled: bool::default(), tracing_search_enabled: bool::default(), otel_otlp_url: String::default(), @@ -3077,7 +3079,6 @@ mod tests { print_key_event: bool::default(), print_key_sql: bool::default(), usage_enabled: bool::default(), - usage_org: String::default(), usage_reporting_mode: String::default(), usage_reporting_url: String::default(), usage_reporting_creds: String::default(), @@ -3162,6 +3163,10 @@ mod tests { query_default_limit: i64::default(), query_partition_by_secs: usize::default(), query_group_base_speed: usize::default(), + file_download_thread_num: usize::default(), + http_keep_alive_disabled: bool::default(), + default_max_query_range_days: i64::default(), + max_dashboard_series: usize::default(), circuit_breaker_enabled: bool::default(), circuit_breaker_watching_window: i64::default(), circuit_breaker_reset_window_num: i64::default(), @@ -3236,11 +3241,6 @@ mod tests { inverted_index_skip_threshold: usize::default(), max_query_range_for_sa: i64::default(), db_text_data_type: String::default(), - wal_write_queue_size: usize::default(), - circuit_breaker_enabled: bool::default(), - circuit_breaker_watching_window: i64::default(), - circuit_breaker_reset_window_num: i64::default(), - circuit_breaker_slow_request_threshold: u64::default(), }, compact: config::Compact { enabled: bool::default(), diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 430a75d110..96575940c5 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -76,7 +76,7 @@ pub const FILE_EXT_TANTIVY_FOLDER: &str = ".mmap"; pub const INDEX_FIELD_NAME_FOR_ALL: &str = "_all"; pub const INDEX_MIN_CHAR_LEN: usize = 3; -pub const QUERY_WITH_NO_LIMIT: i32 = -999; +pub const QUERY_WITH_NO_LIMIT: i64 = -999; pub const MINIMUM_DB_CONNECTIONS: u32 = 2; pub const REQUIRED_DB_CONNECTIONS: u32 = 4; diff --git a/src/config/src/meta/self_reporting/usage.rs b/src/config/src/meta/self_reporting/usage.rs index f3b1c2c030..9351b44859 100644 --- a/src/config/src/meta/self_reporting/usage.rs +++ b/src/config/src/meta/self_reporting/usage.rs @@ -162,27 +162,14 @@ impl std::fmt::Display for UsageEvent { impl From for UsageEvent { fn from(usage: UsageType) -> UsageEvent { - match usage { - UsageType::Bulk - | UsageType::Json - | UsageType::Multi - | UsageType::KinesisFirehose - | UsageType::GCPSubscription - | UsageType::Logs - | UsageType::Traces - | UsageType::Metrics - | UsageType::PrometheusRemoteWrite - | UsageType::JsonMetrics - | UsageType::RUM - | UsageType::EnrichmentTable - | UsageType::Syslog => UsageEvent::Ingestion, - UsageType::Search - | UsageType::SearchAround - | UsageType::SearchTopNValues - | UsageType::MetricSearch - | UsageType::SearchHistory => UsageEvent::Search, - UsageType::Functions => UsageEvent::Functions, - UsageType::Retention => UsageEvent::Other, + if usage.is_ingestion() { + UsageEvent::Ingestion + } else if usage.is_search() { + UsageEvent::Search + } else if usage.is_function() { + UsageEvent::Functions + } else { + UsageEvent::Other } } } @@ -231,6 +218,42 @@ pub enum UsageType { EnrichmentTable, } +impl UsageType { + pub fn is_search(&self) -> bool { + matches!( + self, + UsageType::Search + | UsageType::SearchAround + | UsageType::SearchTopNValues + | UsageType::SearchHistory + | UsageType::MetricSearch + ) + } + + pub fn is_ingestion(&self) -> bool { + matches!( + self, + UsageType::Bulk + | UsageType::Json + | UsageType::Multi + | UsageType::KinesisFirehose + | UsageType::GCPSubscription + | UsageType::Logs + | UsageType::Traces + | UsageType::Metrics + | UsageType::PrometheusRemoteWrite + | UsageType::JsonMetrics + | UsageType::RUM + | UsageType::EnrichmentTable + | UsageType::Syslog + ) + } + + pub fn is_function(&self) -> bool { + matches!(self, UsageType::Functions) + } +} + impl std::fmt::Display for UsageType { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { diff --git a/src/config/src/metrics.rs b/src/config/src/metrics.rs index 9098e030d5..0b42847b2d 100644 --- a/src/config/src/metrics.rs +++ b/src/config/src/metrics.rs @@ -37,7 +37,14 @@ pub static HTTP_INCOMING_REQUESTS: Lazy = Lazy::new(|| { ) .namespace(NAMESPACE) .const_labels(create_const_labels()), - &["endpoint", "status", "organization", "stream_type"], + &[ + "endpoint", + "status", + "organization", + "stream_type", + "search_type", + "search_group", + ], ) .expect("Metric created") }); @@ -49,7 +56,14 @@ pub static HTTP_RESPONSE_TIME: Lazy = Lazy::new(|| { ) .namespace(NAMESPACE) .const_labels(create_const_labels()), - &["endpoint", "status", "organization", "stream_type"], + &[ + "endpoint", + "status", + "organization", + "stream_type", + "search_type", + "search_group", + ], ) .expect("Metric created") }); @@ -63,7 +77,14 @@ pub static GRPC_INCOMING_REQUESTS: Lazy = Lazy::new(|| { ) .namespace(NAMESPACE) .const_labels(create_const_labels()), - &["endpoint", "status", "organization", "stream_type"], + &[ + "endpoint", + "status", + "organization", + "stream_type", + "search_type", + "search_group", + ], ) .expect("Metric created") }); @@ -75,7 +96,14 @@ pub static GRPC_RESPONSE_TIME: Lazy = Lazy::new(|| { ) .namespace(NAMESPACE) .const_labels(create_const_labels()), - &["endpoint", "status", "organization", "stream_type"], + &[ + "endpoint", + "status", + "organization", + "stream_type", + "search_type", + "search_group", + ], ) .expect("Metric created") }); @@ -683,6 +711,15 @@ pub static FILE_LIST_CACHE_HIT_COUNT: Lazy = Lazy::new(|| { }); // Node status metrics +pub static NODE_UP: Lazy = Lazy::new(|| { + IntGaugeVec::new( + Opts::new("node_up", "Node is up") + .namespace(NAMESPACE) + .const_labels(create_const_labels()), + &["version"], + ) + .expect("Metric created") +}); pub static NODE_CPU_TOTAL: Lazy = Lazy::new(|| { IntGaugeVec::new( Opts::new("node_cpu_total", "Total CPU usage") @@ -938,6 +975,9 @@ fn register_metrics(registry: &Registry) { .expect("Metric registered"); // node status metrics + registry + .register(Box::new(NODE_UP.clone())) + .expect("Metric registered"); registry .register(Box::new(NODE_CPU_TOTAL.clone())) .expect("Metric registered"); diff --git a/src/handler/grpc/flight/mod.rs b/src/handler/grpc/flight/mod.rs index aebf1394f9..33841ab030 100644 --- a/src/handler/grpc/flight/mod.rs +++ b/src/handler/grpc/flight/mod.rs @@ -309,10 +309,10 @@ impl Drop for FlightSenderStream { // metrics let time = self.start.elapsed().as_secs_f64(); metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/search/flight/do_get", "200", "", ""]) + .with_label_values(&["/search/flight/do_get", "200", "", "", "", ""]) .observe(time); metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/search/flight/do_get", "200", "", ""]) + .with_label_values(&["/search/flight/do_get", "200", "", "", "", ""]) .inc(); if let Some(defer) = self.defer.take() { diff --git a/src/handler/grpc/request/event.rs b/src/handler/grpc/request/event.rs index d94d497322..d1f4921d48 100644 --- a/src/handler/grpc/request/event.rs +++ b/src/handler/grpc/request/event.rs @@ -90,10 +90,10 @@ impl Event for Eventer { // metrics let time = start.elapsed().as_secs_f64(); metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/event/send_file_list", "200", "", ""]) + .with_label_values(&["/event/send_file_list", "200", "", "", "", ""]) .observe(time); metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/event/send_file_list", "200", "", ""]) + .with_label_values(&["/event/send_file_list", "200", "", "", "", ""]) .inc(); Ok(Response::new(EmptyResponse {})) diff --git a/src/handler/grpc/request/ingest.rs b/src/handler/grpc/request/ingest.rs index dab940b1ad..c6981ad55b 100644 --- a/src/handler/grpc/request/ingest.rs +++ b/src/handler/grpc/request/ingest.rs @@ -167,10 +167,10 @@ impl Ingest for Ingester { // metrics let time = start.elapsed().as_secs_f64(); metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/ingest/inner", "200", "", ""]) + .with_label_values(&["/ingest/inner", "200", "", "", "", ""]) .observe(time); metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/ingest/inner", "200", "", ""]) + .with_label_values(&["/ingest/inner", "200", "", "", "", ""]) .inc(); Ok(Response::new(reply)) diff --git a/src/handler/grpc/request/logs.rs b/src/handler/grpc/request/logs.rs index 443f40ed41..8ec58e2307 100644 --- a/src/handler/grpc/request/logs.rs +++ b/src/handler/grpc/request/logs.rs @@ -71,10 +71,10 @@ impl LogsService for LogsServer { // metrics let time = start.elapsed().as_secs_f64(); metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/otlp/v1/logs", "200", "", ""]) + .with_label_values(&["/otlp/v1/logs", "200", "", "", "", ""]) .observe(time); metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/otlp/v1/logs", "200", "", ""]) + .with_label_values(&["/otlp/v1/logs", "200", "", "", "", ""]) .inc(); Ok(Response::new(ExportLogsServiceResponse { diff --git a/src/handler/grpc/request/metrics/ingester.rs b/src/handler/grpc/request/metrics/ingester.rs index 1747b89ad6..5a3c2d3731 100644 --- a/src/handler/grpc/request/metrics/ingester.rs +++ b/src/handler/grpc/request/metrics/ingester.rs @@ -57,10 +57,10 @@ impl MetricsService for MetricsIngester { // metrics let time = start.elapsed().as_secs_f64(); metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/otlp/v1/metrics", "200", "", ""]) + .with_label_values(&["/otlp/v1/metrics", "200", "", "", "", ""]) .observe(time); metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/otlp/v1/metrics", "200", "", ""]) + .with_label_values(&["/otlp/v1/metrics", "200", "", "", "", ""]) .inc(); return Ok(Response::new(ExportMetricsServiceResponse { partial_success: None, diff --git a/src/handler/grpc/request/metrics/querier.rs b/src/handler/grpc/request/metrics/querier.rs index 6c81533583..63ef14f377 100644 --- a/src/handler/grpc/request/metrics/querier.rs +++ b/src/handler/grpc/request/metrics/querier.rs @@ -47,10 +47,10 @@ impl Metrics for MetricsQuerier { let result = SearchService::grpc::search(req).await.map_err(|err| { let time = start.elapsed().as_secs_f64(); metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/metrics/query", "500", org_id, stream_type]) + .with_label_values(&["/metrics/query", "500", org_id, stream_type, "", ""]) .observe(time); metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/metrics/query", "500", org_id, stream_type]) + .with_label_values(&["/metrics/query", "500", org_id, stream_type, "", ""]) .inc(); let message = if let errors::Error::ErrorCode(code) = err { code.to_json() @@ -62,10 +62,10 @@ impl Metrics for MetricsQuerier { let time = start.elapsed().as_secs_f64(); metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/metrics/query", "200", org_id, stream_type]) + .with_label_values(&["/metrics/query", "200", org_id, stream_type, "", ""]) .observe(time); metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/metrics/query", "200", org_id, stream_type]) + .with_label_values(&["/metrics/query", "200", org_id, stream_type, "", ""]) .inc(); Ok(Response::new(result)) diff --git a/src/handler/grpc/request/traces.rs b/src/handler/grpc/request/traces.rs index 7fd26f6c72..4ab503a065 100644 --- a/src/handler/grpc/request/traces.rs +++ b/src/handler/grpc/request/traces.rs @@ -65,10 +65,10 @@ impl TraceService for TraceServer { // metrics let time = start.elapsed().as_secs_f64(); metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/otlp/v1/traces", "200", "", ""]) + .with_label_values(&["/otlp/v1/traces", "200", "", "", "", ""]) .observe(time); metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/otlp/v1/traces", "200", "", ""]) + .with_label_values(&["/otlp/v1/traces", "200", "", "", "", ""]) .inc(); return Ok(Response::new(ExportTraceServiceResponse { partial_success: None, diff --git a/src/handler/http/request/search/around.rs b/src/handler/http/request/search/around.rs index d55d249ae5..ca84926f9a 100644 --- a/src/handler/http/request/search/around.rs +++ b/src/handler/http/request/search/around.rs @@ -242,7 +242,7 @@ pub(crate) async fn around( resp.cached_ratio = (resp_forward.cached_ratio + resp_backward.cached_ratio) / 2; let time = start.elapsed().as_secs_f64(); - http_report_metrics(start, org_id, stream_type, "200", "_around"); + http_report_metrics(start, org_id, stream_type, "200", "_around", "", ""); let req_stats = RequestStats { records: resp.hits.len() as i64, diff --git a/src/handler/http/request/search/mod.rs b/src/handler/http/request/search/mod.rs index 56360973bb..a08a695259 100644 --- a/src/handler/http/request/search/mod.rs +++ b/src/handler/http/request/search/mod.rs @@ -347,7 +347,19 @@ pub async fn search( match res { Ok(res) => Ok(HttpResponse::Ok().json(res)), Err(err) => { - http_report_metrics(start, &org_id, stream_type, "500", "_search"); + let search_type = req + .search_type + .map(|t| t.to_string()) + .unwrap_or("".to_string()); + http_report_metrics( + start, + &org_id, + stream_type, + "500", + "_search", + &search_type, + "", + ); log::error!("[trace_id {trace_id}] search error: {}", err); Ok(match err { errors::Error::ErrorCode(code) => match code { @@ -456,7 +468,7 @@ pub async fn around_v1( match ret { Ok(res) => Ok(HttpResponse::Ok().json(res)), Err(err) => { - http_report_metrics(start, &org_id, stream_type, "500", "_around"); + http_report_metrics(start, &org_id, stream_type, "500", "_around", "", ""); log::error!("search around error: {:?}", err); Ok(match err { errors::Error::ErrorCode(code) => match code { @@ -575,7 +587,7 @@ pub async fn around_v2( match ret { Ok(res) => Ok(HttpResponse::Ok().json(res)), Err(err) => { - http_report_metrics(start, &org_id, stream_type, "500", "_around"); + http_report_metrics(start, &org_id, stream_type, "500", "_around", "", ""); log::error!("search around error: {:?}", err); Ok(match err { errors::Error::ErrorCode(code) => match code { @@ -912,7 +924,7 @@ async fn values_v1( let resp_search = match search_res { Ok(res) => res, Err(err) => { - http_report_metrics(start, org_id, stream_type, "500", "_values/v1"); + http_report_metrics(start, org_id, stream_type, "500", "_values/v1", "", ""); log::error!("search values error: {:?}", err); return Ok(match err { errors::Error::ErrorCode(code) => match code { @@ -988,7 +1000,7 @@ async fn values_v1( resp.took = start.elapsed().as_millis() as usize; let time = start.elapsed().as_secs_f64(); - http_report_metrics(start, org_id, stream_type, "200", "_values/v1"); + http_report_metrics(start, org_id, stream_type, "200", "_values/v1", "", ""); let req_stats = RequestStats { records: resp.hits.len() as i64, @@ -1105,11 +1117,27 @@ pub async fn search_partition( // do search match search_res { Ok(res) => { - http_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) => { - http_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( @@ -1273,7 +1301,15 @@ pub async fn search_history( let mut search_res = match search_res { Ok(res) => res, Err(err) => { - http_report_metrics(start, &org_id, stream_type, "500", "_search_history"); + http_report_metrics( + start, + &org_id, + stream_type, + "500", + "_search_history", + "", + "", + ); log::error!("[trace_id {}] Search history error : {:?}", trace_id, err); return Ok(match err { errors::Error::ErrorCode(code) => HttpResponse::InternalServerError().json( @@ -1308,7 +1344,15 @@ pub async fn search_history( search_res.trace_id = trace_id.clone(); // report http metrics - http_report_metrics(start, &org_id, stream_type, "200", "_search_history"); + http_report_metrics( + start, + &org_id, + stream_type, + "200", + "_search_history", + "", + "", + ); // prepare usage metrics let time_taken = start.elapsed().as_secs_f64(); diff --git a/src/handler/http/request/search/multi_streams.rs b/src/handler/http/request/search/multi_streams.rs index 4b2cf55820..30aba54702 100644 --- a/src/handler/http/request/search/multi_streams.rs +++ b/src/handler/http/request/search/multi_streams.rs @@ -366,6 +366,8 @@ pub async fn search_multi( "200", &org_id, stream_type.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -374,6 +376,8 @@ pub async fn search_multi( "200", &org_id, stream_type.as_str(), + "", + "", ]) .inc(); res.set_trace_id(trace_id); @@ -458,6 +462,8 @@ pub async fn search_multi( "500", &org_id, stream_type.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -466,6 +472,8 @@ pub async fn search_multi( "500", &org_id, stream_type.as_str(), + "", + "", ]) .inc(); @@ -726,6 +734,8 @@ pub async fn _search_partition_multi( "200", &org_id, stream_type.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -734,6 +744,8 @@ pub async fn _search_partition_multi( "200", &org_id, stream_type.as_str(), + "", + "", ]) .inc(); Ok(HttpResponse::Ok().json(res)) @@ -746,6 +758,8 @@ pub async fn _search_partition_multi( "500", &org_id, stream_type.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -754,6 +768,8 @@ pub async fn _search_partition_multi( "500", &org_id, stream_type.as_str(), + "", + "", ]) .inc(); log::error!("search error: {:?}", err); @@ -889,6 +905,8 @@ pub async fn around_multi( "500", &org_id, stream_type.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -897,6 +915,8 @@ pub async fn around_multi( "500", &org_id, stream_type.as_str(), + "", + "", ]) .inc(); log::error!("multi search around error: {:?}", err); diff --git a/src/handler/http/request/traces/mod.rs b/src/handler/http/request/traces/mod.rs index 43f767492f..fa30ed8cd2 100644 --- a/src/handler/http/request/traces/mod.rs +++ b/src/handler/http/request/traces/mod.rs @@ -322,6 +322,8 @@ pub async fn get_latest_traces( "500", &org_id, stream_type.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -330,6 +332,8 @@ pub async fn get_latest_traces( "500", &org_id, stream_type.as_str(), + "", + "", ]) .inc(); log::error!("get traces latest data error: {:?}", err); @@ -411,6 +415,8 @@ pub async fn get_latest_traces( "500", &org_id, stream_type.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -419,6 +425,8 @@ pub async fn get_latest_traces( "500", &org_id, stream_type.as_str(), + "", + "", ]) .inc(); log::error!("get traces latest data error: {:?}", err); @@ -502,6 +510,8 @@ pub async fn get_latest_traces( "200", &org_id, stream_type.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -510,6 +520,8 @@ pub async fn get_latest_traces( "200", &org_id, stream_type.as_str(), + "", + "", ]) .inc(); diff --git a/src/service/alerts/mod.rs b/src/service/alerts/mod.rs index 58a20bb0c7..1eaea811cf 100644 --- a/src/service/alerts/mod.rs +++ b/src/service/alerts/mod.rs @@ -35,7 +35,7 @@ use config::{ }; use super::promql; -use crate::service::search as SearchService; +use crate::service::{search as SearchService, self_reporting::http_report_metrics}; pub mod alert; pub mod derived_streams; @@ -248,6 +248,7 @@ impl QueryConditionExt for QueryCondition { }; let trace_id = ider::uuid(); + let req_start = std::time::Instant::now(); let resp = if self.multi_time_range.is_some() && !self.multi_time_range.as_ref().unwrap().is_empty() { @@ -391,6 +392,16 @@ impl QueryConditionExt for QueryCondition { // 2. Vec>> - for multi_time_range alert let resp = match resp { Ok(v) => { + // the search request doesn't via cache layer, so need report usage separately + http_report_metrics( + req_start, + org_id, + stream_type, + "200", + "_search", + &SearchEventType::Alerts.to_string(), + "", + ); if v.is_partial { return Err(anyhow::anyhow!("Partial response: {}", v.function_error)); } else { diff --git a/src/service/file_list.rs b/src/service/file_list.rs index 66116ee141..53cefa1b23 100644 --- a/src/service/file_list.rs +++ b/src/service/file_list.rs @@ -196,6 +196,7 @@ pub async fn calculate_files_size(files: &[FileKey]) -> Result { stats.records += file.meta.records; stats.original_size += file.meta.original_size; stats.compressed_size += file.meta.compressed_size; + stats.idx_scan_size += file.meta.index_size; } Ok(stats) } diff --git a/src/service/logs/bulk.rs b/src/service/logs/bulk.rs index a1de94e3c5..74be5ea731 100644 --- a/src/service/logs/bulk.rs +++ b/src/service/logs/bulk.rs @@ -517,6 +517,8 @@ pub async fn ingest( metric_rpt_status_code, org_id, StreamType::Logs.as_str(), + "", + "", ]) .observe(took_time); metrics::HTTP_INCOMING_REQUESTS @@ -525,6 +527,8 @@ pub async fn ingest( metric_rpt_status_code, org_id, StreamType::Logs.as_str(), + "", + "", ]) .inc(); diff --git a/src/service/logs/ingest.rs b/src/service/logs/ingest.rs index 704dae34d2..f09b291d5e 100644 --- a/src/service/logs/ingest.rs +++ b/src/service/logs/ingest.rs @@ -394,6 +394,8 @@ pub async fn ingest( metric_rpt_status_code, org_id, StreamType::Logs.as_str(), + "", + "", ]) .observe(took_time); metrics::HTTP_INCOMING_REQUESTS @@ -402,6 +404,8 @@ pub async fn ingest( metric_rpt_status_code, org_id, StreamType::Logs.as_str(), + "", + "", ]) .inc(); diff --git a/src/service/logs/otlp_grpc.rs b/src/service/logs/otlp_grpc.rs index e7801cb4e6..a6d3e63a75 100644 --- a/src/service/logs/otlp_grpc.rs +++ b/src/service/logs/otlp_grpc.rs @@ -408,6 +408,8 @@ pub async fn handle_grpc_request( metric_rpt_status_code, org_id, StreamType::Logs.as_str(), + "", + "", ]) .observe(took_time); metrics::HTTP_INCOMING_REQUESTS @@ -416,6 +418,8 @@ pub async fn handle_grpc_request( metric_rpt_status_code, org_id, StreamType::Logs.as_str(), + "", + "", ]) .inc(); diff --git a/src/service/logs/otlp_http.rs b/src/service/logs/otlp_http.rs index eeb86cf7bc..67272cf219 100644 --- a/src/service/logs/otlp_http.rs +++ b/src/service/logs/otlp_http.rs @@ -535,6 +535,8 @@ pub async fn logs_json_handler( metric_rpt_status_code, org_id, StreamType::Logs.as_str(), + "", + "", ]) .observe(took_time); metrics::HTTP_INCOMING_REQUESTS @@ -543,6 +545,8 @@ pub async fn logs_json_handler( metric_rpt_status_code, org_id, StreamType::Logs.as_str(), + "", + "", ]) .inc(); diff --git a/src/service/logs/syslog.rs b/src/service/logs/syslog.rs index 5972c0ced4..0d752453e1 100644 --- a/src/service/logs/syslog.rs +++ b/src/service/logs/syslog.rs @@ -372,6 +372,8 @@ pub async fn ingest(msg: &str, addr: SocketAddr) -> Result { metric_rpt_status_code, org_id, StreamType::Logs.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -380,6 +382,8 @@ pub async fn ingest(msg: &str, addr: SocketAddr) -> Result { metric_rpt_status_code, org_id, StreamType::Logs.as_str(), + "", + "", ]) .inc(); diff --git a/src/service/metrics/json.rs b/src/service/metrics/json.rs index 315d4d03b4..15d0a61616 100644 --- a/src/service/metrics/json.rs +++ b/src/service/metrics/json.rs @@ -483,6 +483,8 @@ pub async fn ingest(org_id: &str, body: web::Bytes) -> Result "200", org_id, StreamType::Metrics.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -491,6 +493,8 @@ pub async fn ingest(org_id: &str, body: web::Bytes) -> Result "200", org_id, StreamType::Metrics.as_str(), + "", + "", ]) .inc(); diff --git a/src/service/metrics/otlp.rs b/src/service/metrics/otlp.rs index 688f82cc40..4e9ed32c68 100644 --- a/src/service/metrics/otlp.rs +++ b/src/service/metrics/otlp.rs @@ -580,10 +580,10 @@ pub async fn handle_otlp_request( let time_took = start.elapsed().as_secs_f64(); metrics::HTTP_RESPONSE_TIME - .with_label_values(&[ep, "200", org_id, StreamType::Metrics.as_str()]) + .with_label_values(&[ep, "200", org_id, StreamType::Metrics.as_str(), "", ""]) .observe(time_took); metrics::HTTP_INCOMING_REQUESTS - .with_label_values(&[ep, "200", org_id, StreamType::Metrics.as_str()]) + .with_label_values(&[ep, "200", org_id, StreamType::Metrics.as_str(), "", ""]) .inc(); // only one trigger per request, as it updates etcd diff --git a/src/service/metrics/prom.rs b/src/service/metrics/prom.rs index 5a4ff9c778..d941f3527d 100644 --- a/src/service/metrics/prom.rs +++ b/src/service/metrics/prom.rs @@ -161,6 +161,8 @@ pub async fn remote_write( "200", org_id, StreamType::Metrics.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -169,6 +171,8 @@ pub async fn remote_write( "200", org_id, StreamType::Metrics.as_str(), + "", + "", ]) .inc(); return Ok(()); @@ -263,6 +267,8 @@ pub async fn remote_write( "200", org_id, StreamType::Metrics.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -271,6 +277,8 @@ pub async fn remote_write( "200", org_id, StreamType::Metrics.as_str(), + "", + "", ]) .inc(); return Ok(()); @@ -564,6 +572,8 @@ pub async fn remote_write( "200", org_id, StreamType::Metrics.as_str(), + "", + "", ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS @@ -572,6 +582,8 @@ pub async fn remote_write( "200", org_id, StreamType::Metrics.as_str(), + "", + "", ]) .inc(); diff --git a/src/service/search/cache/mod.rs b/src/service/search/cache/mod.rs index 2cf22ac311..6413426c60 100644 --- a/src/service/search/cache/mod.rs +++ b/src/service/search/cache/mod.rs @@ -296,7 +296,22 @@ pub async fn search( // do search let time = start.elapsed().as_secs_f64(); - http_report_metrics(start, org_id, stream_type, "200", "_search"); + let work_group = get_work_group(work_group_set); + + let search_type = req + .search_type + .map(|t| t.to_string()) + .unwrap_or("".to_string()); + let search_group = work_group.clone().unwrap_or("".to_string()); + http_report_metrics( + start, + org_id, + stream_type, + "200", + "_search", + &search_type, + &search_group, + ); res.set_trace_id(trace_id.to_string()); res.set_local_took(start.elapsed().as_millis() as usize, ext_took_wait); @@ -308,7 +323,6 @@ pub async fn search( res.histogram_interval = Some(c_resp.histogram_interval); } - let work_group = get_work_group(work_group_set); let num_fn = req.query.query_fn.is_some() as u16; let req_stats = RequestStats { records: res.hits.len() as i64, diff --git a/src/service/search/cluster/flight.rs b/src/service/search/cluster/flight.rs index eba5be37e9..20a8b65ba2 100644 --- a/src/service/search/cluster/flight.rs +++ b/src/service/search/cluster/flight.rs @@ -979,7 +979,7 @@ pub async fn get_inverted_index_file_list( req.stream_type = StreamType::Index; query.sql = sql; query.from = 0; - query.size = QUERY_WITH_NO_LIMIT; + query.size = QUERY_WITH_NO_LIMIT as i32; query.track_total_hits = false; query.uses_zo_fn = false; query.query_fn = "".to_string(); diff --git a/src/service/search/datafusion/optimizer/mod.rs b/src/service/search/datafusion/optimizer/mod.rs index ff9f6dae11..88ad092469 100644 --- a/src/service/search/datafusion/optimizer/mod.rs +++ b/src/service/search/datafusion/optimizer/mod.rs @@ -53,7 +53,7 @@ pub mod utils; pub fn generate_optimizer_rules(sql: &Sql) -> Vec> { let cfg = config::get_config(); - let limit = if sql.limit as i32 > config::QUERY_WITH_NO_LIMIT { + let limit = if sql.limit > config::QUERY_WITH_NO_LIMIT { if sql.limit > 0 { Some(sql.limit as usize) } else { diff --git a/src/service/search/grpc/flight.rs b/src/service/search/grpc/flight.rs index 8a3381ea73..8a8cc6f236 100644 --- a/src/service/search/grpc/flight.rs +++ b/src/service/search/grpc/flight.rs @@ -453,10 +453,10 @@ fn collect_stats(files: &[FileKey]) -> ScanStats { let mut scan_stats = ScanStats::new(); scan_stats.files = files.len() as i64; for file in files.iter() { - scan_stats.idx_scan_size += file.meta.index_size; scan_stats.records += file.meta.records; scan_stats.original_size += file.meta.original_size; scan_stats.compressed_size += file.meta.compressed_size; + scan_stats.idx_scan_size += file.meta.index_size; } scan_stats } diff --git a/src/service/search/grpc/storage.rs b/src/service/search/grpc/storage.rs index 58ef222d8c..ac67a24bfc 100644 --- a/src/service/search/grpc/storage.rs +++ b/src/service/search/grpc/storage.rs @@ -191,6 +191,7 @@ pub async fn search( scan_stats.records += file.meta.records; scan_stats.original_size += file.meta.original_size; scan_stats.compressed_size += file.meta.compressed_size; + scan_stats.idx_scan_size += file.meta.index_size; // check schema version let schema_ver_id = match db::schema::filter_schema_version_id( &schema_versions, diff --git a/src/service/search/index.rs b/src/service/search/index.rs index 86e2c4e391..d8d907441d 100644 --- a/src/service/search/index.rs +++ b/src/service/search/index.rs @@ -97,8 +97,8 @@ impl Debug for IndexCondition { pub enum Condition { // field, value Equal(String, String), - Regex(String, String), In(String, Vec), + Regex(String, String), MatchAll(String), FuzzyMatchAll(String, u8), Or(Box, Box), @@ -173,6 +173,12 @@ impl IndexCondition { .collect::, _>>()?, )) } + + pub fn can_remove_filter(&self) -> bool { + self.conditions + .iter() + .all(|condition| condition.can_remove_filter()) + } } impl Condition { @@ -180,8 +186,8 @@ impl Condition { pub fn to_query(&self) -> String { match self { Condition::Equal(field, value) => format!("{}={}", field, value), - Condition::Regex(field, value) => format!("{}=~{}", field, value), Condition::In(field, values) => format!("{} IN ({})", field, values.join(",")), + Condition::Regex(field, value) => format!("{}=~{}", field, value), Condition::MatchAll(value) => format!("{}:{}", INDEX_FIELD_NAME_FOR_ALL, value), Condition::FuzzyMatchAll(value, distance) => format!( "{}:fuzzy({}, {})", @@ -279,10 +285,6 @@ impl Condition { let term = Term::from_field_text(field, value); Box::new(TermQuery::new(term, IndexRecordOption::Basic)) } - Condition::Regex(field, value) => { - let field = schema.get_field(field)?; - Box::new(RegexQuery::from_pattern(value, field)?) - } Condition::In(field, values) => { let field = schema.get_field(field)?; let terms: Vec> = values @@ -294,6 +296,10 @@ impl Condition { .collect(); Box::new(BooleanQuery::union(terms)) } + Condition::Regex(field, value) => { + let field = schema.get_field(field)?; + Box::new(RegexQuery::from_pattern(value, field)?) + } Condition::MatchAll(value) => { let default_field = default_field.ok_or_else(|| { anyhow::anyhow!("There's no FullTextSearch field for match_all() function") @@ -363,10 +369,10 @@ impl Condition { Condition::Equal(field, _) => { fields.insert(field.clone()); } - Condition::Regex(field, _) => { + Condition::In(field, _) => { fields.insert(field.clone()); } - Condition::In(field, _) => { + Condition::Regex(field, _) => { fields.insert(field.clone()); } Condition::MatchAll(_) => { @@ -389,10 +395,10 @@ impl Condition { Condition::Equal(field, _) => { fields.insert(field.clone()); } - Condition::Regex(field, _) => { + Condition::In(field, _) => { fields.insert(field.clone()); } - Condition::In(field, _) => { + Condition::Regex(field, _) => { fields.insert(field.clone()); } Condition::MatchAll(_) => { @@ -422,9 +428,6 @@ impl Condition { let right = get_scalar_value(value, field.data_type())?; Ok(Arc::new(BinaryExpr::new(left, Operator::Eq, right))) } - Condition::Regex(..) => { - unreachable!("Condition::Regex query only support for promql") - } Condition::In(name, values) => { let index = schema.index_of(name).unwrap(); let left = Arc::new(Column::new(name, index)); @@ -435,6 +438,9 @@ impl Condition { .collect::, _>>()?; Ok(Arc::new(InListExpr::new(left, values, false, None))) } + Condition::Regex(..) => { + unreachable!("Condition::Regex query only support for promql") + } Condition::MatchAll(value) => { let value = value .trim_start_matches("re:") // regex @@ -498,6 +504,18 @@ impl Condition { } } } + + pub fn can_remove_filter(&self) -> bool { + match self { + Condition::Equal(..) => true, + Condition::In(..) => true, + Condition::Regex(..) => false, + Condition::MatchAll(v) => is_blank_or_alphanumeric(v), + Condition::FuzzyMatchAll(..) => false, + Condition::Or(left, right) => left.can_remove_filter() && right.can_remove_filter(), + Condition::And(left, right) => left.can_remove_filter() && right.can_remove_filter(), + } + } } // check if function is match_all and only have one argument @@ -630,3 +648,8 @@ fn get_scalar_value(value: &str, data_type: &DataType) -> Result, a _ => unimplemented!(), }) } + +fn is_blank_or_alphanumeric(s: &str) -> bool { + s.chars() + .all(|c| c.is_ascii_whitespace() || c.is_ascii_alphanumeric()) +} diff --git a/src/service/search/sql.rs b/src/service/search/sql.rs index 7453c9f42c..e113a2f51c 100644 --- a/src/service/search/sql.rs +++ b/src/service/search/sql.rs @@ -726,11 +726,16 @@ impl VisitorMut for IndexVisitor { if let Some(expr) = select.selection.as_mut() { let (index, other_expr) = get_index_condition_from_expr(&self.index_fields, expr); self.index_condition = index; + let can_remove_filter = self + .index_condition + .as_ref() + .map(|v| v.can_remove_filter()) + .unwrap_or(true); // make sure all filter in where clause can be used in inverted index - if other_expr.is_none() && select.selection.is_some() { + if other_expr.is_none() && select.selection.is_some() && can_remove_filter { self.can_optimize = true; } - if self.is_remove_filter { + if self.is_remove_filter && can_remove_filter { select.selection = other_expr; } } diff --git a/src/service/self_reporting/mod.rs b/src/service/self_reporting/mod.rs index 5b09f15631..bd7b2ba20a 100644 --- a/src/service/self_reporting/mod.rs +++ b/src/service/self_reporting/mod.rs @@ -81,19 +81,21 @@ pub async fn report_request_usage_stats( num_functions: u16, timestamp: i64, ) { - metrics::INGEST_RECORDS - .with_label_values(&[org_id, stream_type.as_str()]) - .inc_by(stats.records as u64); - metrics::INGEST_BYTES - .with_label_values(&[org_id, stream_type.as_str()]) - .inc_by((stats.size * SIZE_IN_MB) as u64); let event: UsageEvent = usage_type.into(); - let now = DateTime::from_timestamp_micros(timestamp).unwrap(); + if matches!(event, UsageEvent::Ingestion) { + metrics::INGEST_RECORDS + .with_label_values(&[org_id, stream_type.as_str()]) + .inc_by(stats.records as u64); + metrics::INGEST_BYTES + .with_label_values(&[org_id, stream_type.as_str()]) + .inc_by((stats.size * SIZE_IN_MB) as u64); + } if !get_config().common.usage_enabled { return; } + let now = DateTime::from_timestamp_micros(timestamp).unwrap(); let request_body = stats.request_body.unwrap_or(usage_type.to_string()); let user_email = stats.user_email.unwrap_or("".to_owned()); @@ -318,13 +320,29 @@ pub fn http_report_metrics( stream_type: StreamType, code: &str, uri: &str, + search_type: &str, + search_group: &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_type.as_str()]) + .with_label_values(&[ + &uri, + code, + org_id, + stream_type.as_str(), + search_type, + search_group, + ]) .observe(time); metrics::HTTP_INCOMING_REQUESTS - .with_label_values(&[&uri, code, org_id, stream_type.as_str()]) + .with_label_values(&[ + &uri, + code, + org_id, + stream_type.as_str(), + search_type, + search_group, + ]) .inc(); } diff --git a/src/service/traces/mod.rs b/src/service/traces/mod.rs index 7365181315..94d765d688 100644 --- a/src/service/traces/mod.rs +++ b/src/service/traces/mod.rs @@ -547,10 +547,10 @@ pub async fn handle_otlp_request( } metrics::HTTP_RESPONSE_TIME - .with_label_values(&[ep, "200", org_id, StreamType::Traces.as_str()]) + .with_label_values(&[ep, "200", org_id, StreamType::Traces.as_str(), "", ""]) .observe(time); metrics::HTTP_INCOMING_REQUESTS - .with_label_values(&[ep, "200", org_id, StreamType::Traces.as_str()]) + .with_label_values(&[ep, "200", org_id, StreamType::Traces.as_str(), "", ""]) .inc(); format_response(partial_success, req_type) @@ -681,10 +681,10 @@ pub async fn ingest_json( }; metrics::HTTP_RESPONSE_TIME - .with_label_values(&[ep, "200", org_id, StreamType::Traces.as_str()]) + .with_label_values(&[ep, "200", org_id, StreamType::Traces.as_str(), "", ""]) .observe(time); metrics::HTTP_INCOMING_REQUESTS - .with_label_values(&[ep, "200", org_id, StreamType::Traces.as_str()]) + .with_label_values(&[ep, "200", org_id, StreamType::Traces.as_str(), "", ""]) .inc(); format_response(partial_success, req_type)