starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-05 00:09:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf: add auto check if we can remove filter (#6283)

Browse Source
This commit is contained in:
Hengfei Yang
2025-03-17 23:57:49 +08:00
committed by GitHub
parent 37079bec30
commit ef90217770
35 changed files with 348 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/infra/cluster/mod.rs
+3
View File
@@ -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);
src/common/utils/auth_tests.rs
+9 -9
View File
@@ -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(),
src/config/src/config.rs
+1 -1
View File
@@ -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;
src/config/src/meta/self_reporting/usage.rs
+44 -21
View File
@@ -162,27 +162,14 @@ impl std::fmt::Display for UsageEvent {
impl From<UsageType> 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 {
src/config/src/metrics.rs
+44 -4
View File
@@ -37,7 +37,14 @@ pub static HTTP_INCOMING_REQUESTS: Lazy<IntCounterVec> = 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<HistogramVec> = 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<IntCounterVec> = 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<HistogramVec> = 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<IntGaugeVec> = Lazy::new(|| {
});
// Node status metrics
pub static NODE_UP: Lazy<IntGaugeVec> = 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<IntGaugeVec> = 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");
src/handler/grpc/flight/mod.rs
+2 -2
View File
@@ -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() {
src/handler/grpc/request/event.rs
+2 -2
View File
@@ -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 {}))
src/handler/grpc/request/ingest.rs
+2 -2
View File
@@ -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))
src/handler/grpc/request/logs.rs
+2 -2
View File
@@ -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 {
src/handler/grpc/request/metrics/ingester.rs
+2 -2
View File
@@ -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,
src/handler/grpc/request/metrics/querier.rs
+4 -4
View File
@@ -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))
src/handler/grpc/request/traces.rs
+2 -2
View File
@@ -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,
src/handler/http/request/search/around.rs
+1 -1
View File
@@ -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,
src/handler/http/request/search/mod.rs
+53 -9
View File
@@ -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();
src/handler/http/request/search/multi_streams.rs
+20
View File
@@ -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);
src/handler/http/request/traces/mod.rs
+12
View File
@@ -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();
src/service/alerts/mod.rs
+12 -1
View File
@@ -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<Vec<Map<String, Value>>> - 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 {
src/service/file_list.rs
+1
View File
@@ -196,6 +196,7 @@ pub async fn calculate_files_size(files: &[FileKey]) -> Result<ScanStats> {
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)
}
src/service/logs/bulk.rs
+4
View File
@@ -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();
src/service/logs/ingest.rs
+4
View File
@@ -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();
src/service/logs/otlp_grpc.rs
+4
View File
@@ -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();
src/service/logs/otlp_http.rs
+4
View File
@@ -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();
src/service/logs/syslog.rs
+4
View File
@@ -372,6 +372,8 @@ pub async fn ingest(msg: &str, addr: SocketAddr) -> Result<HttpResponse> {
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<HttpResponse> {
metric_rpt_status_code,
org_id,
StreamType::Logs.as_str(),
"",
"",
])
.inc();
src/service/metrics/json.rs
+4
View File
@@ -483,6 +483,8 @@ pub async fn ingest(org_id: &str, body: web::Bytes) -> Result<IngestionResponse>
"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<IngestionResponse>
"200",
org_id,
StreamType::Metrics.as_str(),
"",
"",
])
.inc();
src/service/metrics/otlp.rs
+2 -2
View File
@@ -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
src/service/metrics/prom.rs
+12
View File
@@ -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();
src/service/search/cache/mod.rs
Vendored
+16 -2
View File
@@ -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,
src/service/search/cluster/flight.rs
+1 -1
View File
@@ -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();
src/service/search/datafusion/optimizer/mod.rs
+1 -1
View File
@@ -53,7 +53,7 @@ pub mod utils;
pub fn generate_optimizer_rules(sql: &Sql) -> Vec<Arc<dyn OptimizerRule + Send + Sync>> {
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 {
src/service/search/grpc/flight.rs
+1 -1
View File
@@ -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
}
src/service/search/grpc/storage.rs
+1
View File
@@ -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,
src/service/search/index.rs
+36 -13
View File
@@ -97,8 +97,8 @@ impl Debug for IndexCondition {
pub enum Condition {
// field, value
Equal(String, String),
Regex(String, String),
In(String, Vec<String>),
Regex(String, String),
MatchAll(String),
FuzzyMatchAll(String, u8),
Or(Box<Condition>, Box<Condition>),
@@ -173,6 +173,12 @@ impl IndexCondition {
.collect::<Result<Vec<_>, _>>()?,
))
}
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<Box<dyn Query>> = 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::<Result<Vec<_>, _>>()?;
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<Arc<Literal>, a
_ => unimplemented!(),
})
}
fn is_blank_or_alphanumeric(s: &str) -> bool {
s.chars()
.all(|c| c.is_ascii_whitespace() || c.is_ascii_alphanumeric())
}
src/service/search/sql.rs
+7 -2
View File
@@ -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;
}
}
src/service/self_reporting/mod.rs
+27 -9
View File
@@ -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();
}
src/service/traces/mod.rs
+4 -4
View File
@@ -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)