From 4eebe254514a186eef860bfb780cb62b75cfb3ec Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Thu, 17 Sep 2026 04:38:59 +0000 Subject: [PATCH] fix: defer expensive fallbacks and filter null VRL results (#14559) --- .../grpc/src/handler/grpc/request/ingest.rs | 55 +++++-- src/api/management/src/request/ai/chat.rs | 5 +- src/api/management/src/request/ai/toolsets.rs | 2 +- .../src/request/organization/org.rs | 6 +- src/api/search/src/search/around.rs | 8 +- src/api/search/src/search/mod.rs | 4 +- src/config/src/meta/alerts/mod.rs | 5 +- src/config/src/meta/sql.rs | 4 +- src/config/src/utils/download_utils.rs | 10 +- src/config/src/utils/sysinfo/os.rs | 6 +- src/core/src/alerts/alert.rs | 8 +- src/core/src/functions.rs | 3 +- src/core/src/incidents.rs | 17 ++- src/core/src/logs/ingest.rs | 19 ++- src/core/src/logs/otlp.rs | 2 +- src/core/src/metrics/json.rs | 11 +- src/core/src/metrics/prom.rs | 2 +- src/core/src/pipeline/batch_execution.rs | 4 +- src/core/src/self_reporting/search.rs | 2 +- src/core/src/slo/job.rs | 2 +- src/core/src/traces/inferred.rs | 144 ++++++++++++++++-- src/core/src/users.rs | 2 +- src/core/src/workflows/mod.rs | 12 +- src/db/src/alerts/alert.rs | 2 +- src/db/src/alerts/destinations.rs | 2 +- .../src/enrichment_table/mod.rs | 2 +- .../src/enrichment_table/url_processor.rs | 4 +- src/infra/src/coordinator/alerts.rs | 2 +- src/infra/src/table/ratelimit.rs | 12 +- src/metrics_index/src/pruner.rs | 14 +- src/search/src/tantivy/mod.rs | 2 +- src/search_service/src/cache/mod.rs | 63 +++++++- src/search_service/src/cluster/http.rs | 2 +- src/search_service/src/file_list.rs | 3 +- src/search_service/src/grpc/flight.rs | 9 +- src/search_service/src/lib.rs | 31 +++- .../src/partition/sql_context.rs | 3 +- src/stream/src/lib.rs | 2 +- src/super_cluster_queue/src/pipelines.rs | 8 +- src/super_cluster_queue/src/ratelimit.rs | 2 +- src/synthetics/src/status_pages.rs | 2 +- src/usage_reporting/src/lib.rs | 2 +- 42 files changed, 369 insertions(+), 131 deletions(-) diff --git a/src/api/grpc/src/handler/grpc/request/ingest.rs b/src/api/grpc/src/handler/grpc/request/ingest.rs index c5bb4b11c0..3047b4cd4c 100644 --- a/src/api/grpc/src/handler/grpc/request/ingest.rs +++ b/src/api/grpc/src/handler/grpc/request/ingest.rs @@ -118,17 +118,7 @@ impl Ingest for Ingester { } } StreamType::EnrichmentTables => { - let json_records: Vec> = - json::from_slice(&in_data.data).unwrap_or({ - let vec_value: Vec = json::from_slice(&in_data.data).unwrap(); - vec_value - .into_iter() - .filter_map(|v| match v { - json::Value::Object(map) => Some(map), - _ => None, - }) - .collect() - }); + let json_records = parse_enrichment_records(&in_data.data); let append_data = match req.metadata { Some(metadata) => metadata .data @@ -217,6 +207,19 @@ fn ok_reply() -> IngestionResponse { } } +fn parse_enrichment_records(data: &[u8]) -> Vec> { + json::from_slice(data).unwrap_or_else(|_| { + let vec_value: Vec = json::from_slice(data).unwrap(); + vec_value + .into_iter() + .filter_map(|v| match v { + json::Value::Object(map) => Some(map), + _ => None, + }) + .collect() + }) +} + /// The proto has only `status_code` + `message`, so `207` carries the partial-failure JSON. fn encode_metrics_reply(resp: &ingestion_common::IngestionResponse) -> IngestionResponse { if resp.code != 200 { @@ -264,6 +267,36 @@ mod tests { resp } + #[test] + fn test_parse_enrichment_records_object_array() { + let records = + parse_enrichment_records(br#"[{"id": 1}, {"id": 2, "nested": {"ok": true}}]"#); + assert_eq!( + json::to_value(records).unwrap(), + json::json!([ + {"id": 1}, {"id": 2, "nested": {"ok": true}} + ]) + ); + assert!(parse_enrichment_records(b"[]").is_empty()); + } + + #[test] + fn test_parse_enrichment_records_mixed_array() { + let records = + parse_enrichment_records(br#"[null, {"id": 1}, 2, "text", false, [], {"id": 2}]"#); + assert_eq!( + json::to_value(records).unwrap(), + json::json!([{"id": 1}, {"id": 2}]) + ); + assert!(parse_enrichment_records(b"[null, 1, false]").is_empty()); + } + + #[test] + #[should_panic] + fn test_parse_enrichment_records_rejects_non_array() { + parse_enrichment_records(br#"{"id": 1}"#); + } + #[test] fn test_encode_metrics_reply_rejection_keeps_code() { let reply = encode_metrics_reply(&metrics_resp(503, 0, Some("blocked"))); diff --git a/src/api/management/src/request/ai/chat.rs b/src/api/management/src/request/ai/chat.rs index 45042413a8..edc68d587c 100644 --- a/src/api/management/src/request/ai/chat.rs +++ b/src/api/management/src/request/ai/chat.rs @@ -623,7 +623,8 @@ pub async fn chat_stream(Path(org_id): Path, in_req: axum::extract::Requ user_id, forward_headers .get(X_O2_ASSISTANT_SESSION_ID.as_str()) - .unwrap_or(&"none".to_string()) + .map(String::as_str) + .unwrap_or("none") ); // Parse JSON body @@ -1123,7 +1124,7 @@ pub async fn confirm_action( StatusCode::OK, axum::Json( serde_json::from_str::(&response_body) - .unwrap_or(serde_json::json!({"ok": true})), + .unwrap_or_else(|_| serde_json::json!({"ok": true})), ), ) .into_response() diff --git a/src/api/management/src/request/ai/toolsets.rs b/src/api/management/src/request/ai/toolsets.rs index b9bccc77d6..50de4a5368 100644 --- a/src/api/management/src/request/ai/toolsets.rs +++ b/src/api/management/src/request/ai/toolsets.rs @@ -245,7 +245,7 @@ pub async fn update( .data .as_ref() .map(|v| serde_json::to_string(v).unwrap()) - .or(existing.data.clone()); + .or_else(|| existing.data.clone()); let updated = OrgToolset { id: existing.id.clone(), diff --git a/src/api/management/src/request/organization/org.rs b/src/api/management/src/request/organization/org.rs index 0ee7e709cb..aa8146ffbd 100644 --- a/src/api/management/src/request/organization/org.rs +++ b/src/api/management/src/request/organization/org.rs @@ -92,7 +92,8 @@ pub async fn organizations( let limit = query .get("page_size") - .unwrap_or(&"100".to_string()) + .map(String::as_str) + .unwrap_or("100") .parse::() .ok(); let is_root_user = is_root_user(user_id); @@ -190,7 +191,8 @@ pub async fn all_organizations( let mut org_names = HashSet::new(); let limit = query .get("page_size") - .unwrap_or(&"100".to_string()) + .map(String::as_str) + .unwrap_or("100") .parse::() .ok(); diff --git a/src/api/search/src/search/around.rs b/src/api/search/src/search/around.rs index 3dba4de50f..814c1f56b8 100644 --- a/src/api/search/src/search/around.rs +++ b/src/api/search/src/search/around.rs @@ -146,8 +146,8 @@ pub(crate) async fn around( .unwrap(); // search forward - let fw_sql = - check_or_add_order_by_timestamp(&around_sql, false).unwrap_or(around_sql.to_string()); + let fw_sql = check_or_add_order_by_timestamp(&around_sql, false) + .unwrap_or_else(|_| around_sql.to_string()); let fw_sql = config::utils::query_select_utils::replace_o2_custom_patterns(&fw_sql).unwrap_or(fw_sql); @@ -187,8 +187,8 @@ pub(crate) async fn around( .await?; // search backward - let bw_sql = - check_or_add_order_by_timestamp(&around_sql, true).unwrap_or(around_sql.to_string()); + let bw_sql = check_or_add_order_by_timestamp(&around_sql, true) + .unwrap_or_else(|_| around_sql.to_string()); let bw_sql = config::utils::query_select_utils::replace_o2_custom_patterns(&bw_sql).unwrap_or(bw_sql); let req = config::meta::search::Request { diff --git a/src/api/search/src/search/mod.rs b/src/api/search/src/search/mod.rs index 6697e57351..13b05701da 100644 --- a/src/api/search/src/search/mod.rs +++ b/src/api/search/src/search/mod.rs @@ -849,7 +849,7 @@ pub async fn build_search_request_per_field( let schema = infra::schema::get(org_id, stream_name, stream_type) .await - .unwrap_or(Schema::empty()); + .unwrap_or_else(|_| Schema::empty()); let fields = req .fields .iter() @@ -1161,7 +1161,7 @@ async fn values_inner( // skip fields which aren't part of the schema let schema = infra::schema::get(org_id, stream_name, stream_type) .await - .unwrap_or(Schema::empty()); + .unwrap_or_else(|_| Schema::empty()); let mut query_results = Vec::with_capacity(fields.len()); let sql_where = if where_str.is_empty() { diff --git a/src/config/src/meta/alerts/mod.rs b/src/config/src/meta/alerts/mod.rs index c87728e299..41ed05a7cf 100644 --- a/src/config/src/meta/alerts/mod.rs +++ b/src/config/src/meta/alerts/mod.rs @@ -147,9 +147,8 @@ impl TriggerCondition { _ => 0, }; let start_utc = start_from.map_or(Ok(Utc::now()), |from| { - chrono::DateTime::::from_timestamp_micros(from).ok_or(anyhow::anyhow!( - "Error converting start_from value to timestamp" - )) + chrono::DateTime::::from_timestamp_micros(from) + .ok_or_else(|| anyhow::anyhow!("Error converting start_from value to timestamp")) })?; if self.frequency_type == FrequencyType::Cron { diff --git a/src/config/src/meta/sql.rs b/src/config/src/meta/sql.rs index cdcb983f46..eaae07b9ec 100644 --- a/src/config/src/meta/sql.rs +++ b/src/config/src/meta/sql.rs @@ -101,7 +101,7 @@ pub fn resolve_stream_names(sql: &str) -> Result, anyhow::Error> { let dialect = &PostgreSqlDialect {}; let statement = DFParser::parse_sql_with_dialect(sql, dialect)? .pop_back() - .ok_or(anyhow::anyhow!("Failed to parse sql"))?; + .ok_or_else(|| anyhow::anyhow!("Failed to parse sql"))?; let (table_refs, _) = resolve_table_references(&statement, true)?; let mut tables = Vec::new(); for table in table_refs { @@ -114,7 +114,7 @@ pub fn resolve_stream_names_with_type(sql: &str) -> Result, let dialect = &PostgreSqlDialect {}; let statement = DFParser::parse_sql_with_dialect(sql, dialect)? .pop_back() - .ok_or(anyhow::anyhow!("Failed to parse sql"))?; + .ok_or_else(|| anyhow::anyhow!("Failed to parse sql"))?; let (table_refs, _) = resolve_table_references(&statement, true)?; let mut tables = Vec::new(); for table in table_refs { diff --git a/src/config/src/utils/download_utils.rs b/src/config/src/utils/download_utils.rs index 2ba31bb8c7..f17e382c35 100644 --- a/src/config/src/utils/download_utils.rs +++ b/src/config/src/utils/download_utils.rs @@ -26,23 +26,23 @@ pub async fn download_file(client: &Client, url: &str, path: &str) -> Result<(), .get(url) .send() .await - .or(Err(format!("Failed to GET from '{url}'")))?; + .map_err(|_| format!("Failed to GET from '{url}'"))?; let total_size = res .content_length() - .ok_or(format!("Failed to get content length from '{url}'"))?; + .ok_or_else(|| format!("Failed to get content length from '{url}'"))?; // download chunks let mut file = File::create(path) .await - .or(Err(format!("Failed to create file '{path}'")))?; + .map_err(|_| format!("Failed to create file '{path}'"))?; let mut downloaded: u64 = 0; let mut stream = res.bytes_stream(); while let Some(item) = stream.next().await { - let chunk = item.or(Err("Error while downloading file".to_string()))?; + let chunk = item.map_err(|_| "Error while downloading file".to_string())?; file.write_all(&chunk) .await - .or(Err("Error while writing to file".to_string()))?; + .map_err(|_| "Error while writing to file".to_string())?; let new = min(downloaded + (chunk.len() as u64), total_size); downloaded = new; } diff --git a/src/config/src/utils/sysinfo/os.rs b/src/config/src/utils/sysinfo/os.rs index 70856cba5d..d98f704bf7 100644 --- a/src/config/src/utils/sysinfo/os.rs +++ b/src/config/src/utils/sysinfo/os.rs @@ -14,15 +14,15 @@ // along with this program. If not, see . pub fn get_os_name() -> String { - sysinfo::System::name().unwrap_or("unknown".to_string()) + sysinfo::System::name().unwrap_or_else(|| "unknown".to_string()) } pub fn get_os_version() -> String { - sysinfo::System::os_version().unwrap_or("unknown".to_string()) + sysinfo::System::os_version().unwrap_or_else(|| "unknown".to_string()) } pub fn get_hostname() -> String { - sysinfo::System::host_name().unwrap_or("unknown".to_string()) + sysinfo::System::host_name().unwrap_or_else(|| "unknown".to_string()) } #[cfg(target_os = "linux")] diff --git a/src/core/src/alerts/alert.rs b/src/core/src/alerts/alert.rs index 69397f5ae1..88dd71caf5 100644 --- a/src/core/src/alerts/alert.rs +++ b/src/core/src/alerts/alert.rs @@ -2479,10 +2479,10 @@ impl AlertExt for Alert { if !self.workflows.is_empty() { let data: Vec<_> = rows.iter().map(|v| Value::Object(v.clone())).collect(); - let source_id = self - .id - .as_ref() - .map_or(format!("{}/{}", self.org_id, self.name), |v| v.to_string()); + let source_id = self.id.as_ref().map_or_else( + || format!("{}/{}", self.org_id, self.name), + |v| v.to_string(), + ); let metadata: HashMap = vec![ ("org_id", self.org_id.clone().into()), diff --git a/src/core/src/functions.rs b/src/core/src/functions.rs index 54b78f6575..570e83a6a9 100644 --- a/src/core/src/functions.rs +++ b/src/core/src/functions.rs @@ -460,7 +460,8 @@ pub async fn delete_function(org_id: &str, fn_name: &str) -> Result<(), Function } let pipeline_dep = get_dependencies(org_id, fn_name).await; if !pipeline_dep.is_empty() { - let pipeline_data = serde_json::to_string(&pipeline_dep).unwrap_or("[]".to_string()); + let pipeline_data = + serde_json::to_string(&pipeline_dep).unwrap_or_else(|_| "[]".to_string()); return Err(FunctionDeleteError::PipelineDependencies(format!( "Warning: Function '{}' has {} pipeline dependencies. Please remove these pipelines first: {}", fn_name, diff --git a/src/core/src/incidents.rs b/src/core/src/incidents.rs index 6c1ca630a6..ea40b641d9 100644 --- a/src/core/src/incidents.rs +++ b/src/core/src/incidents.rs @@ -29,7 +29,7 @@ async fn get_event_metadata( ) -> Result, anyhow::Error> { let incident = get_incident_with_alerts(org_id, incident_id) .await? - .ok_or(anyhow::anyhow!("incident with given id not found"))?; + .ok_or_else(|| anyhow::anyhow!("incident with given id not found"))?; let mut metadata: Vec<(&str, Value)> = Vec::new(); @@ -97,7 +97,10 @@ async fn get_event_metadata( IncidentEventType::Resolved { user_id } => { metadata.push(("event_type", "resolved".into())); - metadata.push(("user_id", user_id.unwrap_or("system".to_string()).into())); + metadata.push(( + "user_id", + user_id.unwrap_or_else(|| "system".to_string()).into(), + )); } IncidentEventType::Reopened { user_id, reason } => { @@ -120,8 +123,14 @@ async fn get_event_metadata( } IncidentEventType::AssignmentChanged { from, to } => { metadata.push(("event_type", "assignment_changed".into())); - metadata.push(("from_user", from.unwrap_or("unknown".to_string()).into())); - metadata.push(("to_user", to.unwrap_or("unknown".to_string()).into())); + metadata.push(( + "from_user", + from.unwrap_or_else(|| "unknown".to_string()).into(), + )); + metadata.push(( + "to_user", + to.unwrap_or_else(|| "unknown".to_string()).into(), + )); } IncidentEventType::Comment { user_id, comment } => { diff --git a/src/core/src/logs/ingest.rs b/src/core/src/logs/ingest.rs index 6e36518b30..191237d8f8 100644 --- a/src/core/src/logs/ingest.rs +++ b/src/core/src/logs/ingest.rs @@ -1045,7 +1045,7 @@ fn deserialize_aws_record_from_vec(data: Vec, request_id: &str) -> Result, request_id: &str) -> Result>() diff --git a/src/core/src/logs/otlp.rs b/src/core/src/logs/otlp.rs index 25d914cee2..0f3c60800e 100644 --- a/src/core/src/logs/otlp.rs +++ b/src/core/src/logs/otlp.rs @@ -148,7 +148,7 @@ pub async fn handle_request( // check stream let stream_name = in_stream_name .map(|name| format_stream_name(name.to_string())) - .unwrap_or("default".to_string()); + .unwrap_or_else(|| "default".to_string()); check_ingestion_allowed(org_id, StreamType::Logs, Some(&stream_name)).await?; let cfg = get_config(); diff --git a/src/core/src/metrics/json.rs b/src/core/src/metrics/json.rs index 3f826980a9..e359849387 100644 --- a/src/core/src/metrics/json.rs +++ b/src/core/src/metrics/json.rs @@ -112,7 +112,10 @@ async fn buffer_record( }; let stream_name = match stream_name { Some(name) => name.to_string(), - None => match record.get(NAME_LABEL).ok_or(anyhow!("missing __name__"))? { + None => match record + .get(NAME_LABEL) + .ok_or_else(|| anyhow!("missing __name__"))? + { json::Value::String(s) => format_stream_name(s.to_string()), _ => { return Err(anyhow::anyhow!("invalid __name__, need to be string")); @@ -241,7 +244,11 @@ fn finish_rows( let first_type = records.first().map(|(_, metric_type)| *metric_type); let mut rows = Vec::with_capacity(records.len()); for (mut record, _) in records { - let value = parse_metric_value(record.get(VALUE_LABEL).ok_or(anyhow!("missing value"))?)?; + let value = parse_metric_value( + record + .get(VALUE_LABEL) + .ok_or_else(|| anyhow!("missing value"))?, + )?; if let Some(existing) = record.get_mut(VALUE_LABEL) { *existing = value; } diff --git a/src/core/src/metrics/prom.rs b/src/core/src/metrics/prom.rs index 2f2cf58276..80a21afdf5 100644 --- a/src/core/src/metrics/prom.rs +++ b/src/core/src/metrics/prom.rs @@ -148,7 +148,7 @@ pub async fn remote_write( let metric_name = format_stream_name(item.metric_family_name.to_string()); let schema = infra::schema::get(org_id, &metric_name, StreamType::Metrics) .await - .unwrap_or(Schema::empty()); + .unwrap_or_else(|_| Schema::empty()); if schema.metadata().contains_key(METADATA_LABEL) { // already has metadata, skip continue; diff --git a/src/core/src/pipeline/batch_execution.rs b/src/core/src/pipeline/batch_execution.rs index 7eb42ac110..8cde5b51d7 100644 --- a/src/core/src/pipeline/batch_execution.rs +++ b/src/core/src/pipeline/batch_execution.rs @@ -1126,7 +1126,7 @@ impl ExecutablePipeline { .filter_map(|exec_node| { if exec_node.children.is_empty() { if let NodeData::Stream(stream_params) = &exec_node.node_data { - (!stream_params.stream_name.contains("{")).then_some(stream_params.clone()) + (!stream_params.stream_name.contains("{")).then(|| stream_params.clone()) } else { None } @@ -1966,7 +1966,7 @@ async fn process_function_node( let stream_name = metadata .stream_name .clone() - .unwrap_or("pipeline".to_string()); + .unwrap_or_else(|| "pipeline".to_string()); let mut result_array_records = Vec::new(); while let Some(pipeline_item) = channels.receiver.recv().await { channels diff --git a/src/core/src/self_reporting/search.rs b/src/core/src/self_reporting/search.rs index 55e2e1f0ca..2f2f77be07 100644 --- a/src/core/src/self_reporting/search.rs +++ b/src/core/src/self_reporting/search.rs @@ -104,7 +104,7 @@ pub async fn get_usage( pub async fn get_license_usage_data_from_node(node_id: String) -> Result { let node = get_node_by_uuid(&node_id) .await - .ok_or(anyhow::anyhow!("node {node_id} not found"))?; + .ok_or_else(|| anyhow::anyhow!("node {node_id} not found"))?; let node = std::sync::Arc::new(node) as std::sync::Arc; let task = tokio::task::spawn(async move { diff --git a/src/core/src/slo/job.rs b/src/core/src/slo/job.rs index 0a4a1694a4..2d9697098f 100644 --- a/src/core/src/slo/job.rs +++ b/src/core/src/slo/job.rs @@ -416,7 +416,7 @@ pub fn promql_rows( let slice_start = t_micros / 1_000_000 - slice_interval_secs; let e = acc .entry((slice_start, key.clone())) - .or_insert((0.0, labels.clone())); + .or_insert_with(|| (0.0, labels.clone())); e.0 += value; } } diff --git a/src/core/src/traces/inferred.rs b/src/core/src/traces/inferred.rs index b986a21398..cc63a1f66b 100644 --- a/src/core/src/traces/inferred.rs +++ b/src/core/src/traces/inferred.rs @@ -186,18 +186,20 @@ where .filter(|host| !host.is_empty() && !is_ip_address(host)) }; - let fallback_host = host_attr("server.address") - .or_else(|| host_attr("net.peer.name")) - .or_else(|| host_attr("http.host")) - .or_else(|| { - attr("url.full") - .or_else(|| attr("http.url")) - .and_then(|u| url::Url::parse(&u).ok()) - .and_then(|u| match u.host() { - Some(url::Host::Domain(d)) => Some(d.to_string()), - _ => None, // IP hosts are redacted, same as host_attr - }) - }); + let fallback_host = || { + host_attr("server.address") + .or_else(|| host_attr("net.peer.name")) + .or_else(|| host_attr("http.host")) + .or_else(|| { + attr("url.full") + .or_else(|| attr("http.url")) + .and_then(|u| url::Url::parse(&u).ok()) + .and_then(|u| match u.host() { + Some(url::Host::Domain(d)) => Some(d.to_string()), + _ => None, // IP hosts are redacted, same as host_attr + }) + }) + }; // Explicit peer.service overrides every naming rule. let peer_service = attr("peer.service"); @@ -208,7 +210,7 @@ where if db_system.is_some() || db_name.is_some() { let name = peer_service .or(db_name) - .or(fallback_host) + .or_else(fallback_host) .or_else(|| db_system.clone())?; return Some(InferredService { name, @@ -223,7 +225,7 @@ where if messaging_system.is_some() || destination.is_some() { let name = peer_service .or(destination) - .or(fallback_host) + .or_else(fallback_host) .or_else(|| messaging_system.clone())?; return Some(InferredService { name, @@ -238,7 +240,7 @@ where if rpc_system.is_some() || rpc_service.is_some() { let name = peer_service .or(rpc_service) - .or(fallback_host) + .or_else(fallback_host) .or_else(|| rpc_system.clone())?; return Some(InferredService { name, @@ -248,7 +250,7 @@ where } // external (http or generic network peer) - let name = peer_service.or(fallback_host)?; + let name = peer_service.or_else(fallback_host)?; let is_http = attr("http.request.method") .or_else(|| attr("http.method")) .or_else(|| attr("url.full")) @@ -450,6 +452,116 @@ mod tests { peer_keys(SPAN_KIND_CLIENT, attrs).and_then(|keys| keys.key) } + #[test] + fn test_explicit_names_skip_host_lookup_and_preserve_classification() { + for (system_key, system, name_key, service_type) in [ + ( + "db.system", + "postgresql", + "db.namespace", + INFER_TYPE_DATABASE, + ), + ( + "messaging.system", + "kafka", + "messaging.destination.name", + INFER_TYPE_QUEUE, + ), + ("rpc.system", "grpc", "rpc.service", INFER_TYPE_RPC), + ] { + for explicit_peer in [false, true] { + let result = derive_inferred_service(SPAN_KIND_CLIENT, |key| { + assert!( + !matches!( + key, + "server.address" + | "net.peer.name" + | "http.host" + | "url.full" + | "http.url" + ), + "unexpected fallback lookup: {key}" + ); + if key == system_key { + Some(system.to_string()) + } else if key == name_key { + Some("named-dependency".to_string()) + } else if explicit_peer && key == "peer.service" { + Some("explicit-peer".to_string()) + } else { + None + } + }) + .unwrap(); + assert_eq!( + result.name, + if explicit_peer { + "explicit-peer" + } else { + "named-dependency" + } + ); + assert_eq!(result.service_type, service_type); + assert_eq!(result.system.as_deref(), Some(system)); + } + } + } + + #[test] + fn test_external_peer_skips_host_but_keeps_http_classification() { + let result = derive_inferred_service(SPAN_KIND_CLIENT, |key| { + assert!( + !matches!(key, "server.address" | "net.peer.name" | "http.host"), + "unexpected host lookup: {key}" + ); + match key { + "peer.service" => Some("payments".to_string()), + "url.full" => Some("https://example.com/payments".to_string()), + _ => None, + } + }) + .unwrap(); + assert_eq!(result.name, "payments"); + assert_eq!(result.service_type, INFER_TYPE_EXTERNAL); + assert_eq!(result.system.as_deref(), Some("http")); + } + + #[test] + fn test_host_fallback_preserves_precedence() { + for (attrs, name) in [ + ( + vec![ + ("server.address", "primary:443"), + ("net.peer.name", "secondary"), + ("url.full", "https://url-host/path"), + ], + "primary", + ), + ( + vec![ + ("server.address", "127.0.0.1"), + ("net.peer.name", "secondary:80"), + ("url.full", "https://url-host/path"), + ], + "secondary", + ), + ( + vec![ + ("server.address", "127.0.0.1"), + ("url.full", "https://url-host/path"), + ], + "url-host", + ), + ] { + let mut attrs = attrs; + attrs.push(("db.system", "postgresql")); + let result = derive(SPAN_KIND_CLIENT, &attrs).unwrap(); + assert_eq!(result.name, name); + assert_eq!(result.service_type, INFER_TYPE_DATABASE); + assert_eq!(result.system.as_deref(), Some("postgresql")); + } + } + #[test] fn test_database_with_db_name() { let result = derive( diff --git a/src/core/src/users.rs b/src/core/src/users.rs index 68a1f15ee2..c93cf8bb59 100644 --- a/src/core/src/users.rs +++ b/src/core/src/users.rs @@ -1222,7 +1222,7 @@ pub async fn list_user_invites(user_id: &str, only_pending: bool) -> Result Result { let workflow = get_workflow_by_id(org_id, id) .await? - .ok_or(anyhow::anyhow!("workflow with given id not found"))?; + .ok_or_else(|| anyhow::anyhow!("workflow with given id not found"))?; if !workflow.enabled { return Ok(WorkflowExecutionStatus::Success); @@ -942,7 +942,7 @@ pub async fn retry_run( ) -> Result { let workflow = workflows::get_by_org_wid(org_id, wid) .await? - .ok_or(anyhow::anyhow!("workflow with given id not found"))?; + .ok_or_else(|| anyhow::anyhow!("workflow with given id not found"))?; let mut start_id = None; for node in &workflow.nodes { @@ -992,9 +992,11 @@ pub async fn retry_run( let node_id = from_node.as_ref().unwrap_or(&start_id); - let inputs = ip_map.input_map.remove(node_id).ok_or(anyhow::anyhow!( - "node id {node_id} does not have any associated input data in the stored inputs" - ))?; + let inputs = ip_map.input_map.remove(node_id).ok_or_else(|| { + anyhow::anyhow!( + "node id {node_id} does not have any associated input data in the stored inputs" + ) + })?; let start_time = chrono::Utc::now().timestamp_micros(); let res = executable diff --git a/src/db/src/alerts/alert.rs b/src/db/src/alerts/alert.rs index 6c9716f6ee..8eb66d0d21 100644 --- a/src/db/src/alerts/alert.rs +++ b/src/db/src/alerts/alert.rs @@ -582,7 +582,7 @@ pub fn cache_alert_key(org: &str, alert_id: &str) -> String { /// Returns the key used to schedule a trigger for the alert. pub fn scheduler_key(alert_id: Option) -> String { - alert_id.map_or(DEFAULT_FOLDER.to_string(), |id| id.to_string()) + alert_id.map_or_else(|| DEFAULT_FOLDER.to_string(), |id| id.to_string()) } #[cfg(test)] diff --git a/src/db/src/alerts/destinations.rs b/src/db/src/alerts/destinations.rs index 1c4d039671..bee359004a 100644 --- a/src/db/src/alerts/destinations.rs +++ b/src/db/src/alerts/destinations.rs @@ -234,7 +234,7 @@ pub(super) fn parse_event_key<'a>( ) -> Result<(&'a str, &'a str), anyhow::Error> { let item_key = event_key .strip_prefix(prefix) - .ok_or(anyhow::anyhow!("event key missing prefix"))?; + .ok_or_else(|| anyhow::anyhow!("event key missing prefix"))?; let mut keys = item_key.split('/'); let org_id = keys .next() diff --git a/src/enrichment_data/src/enrichment_table/mod.rs b/src/enrichment_data/src/enrichment_table/mod.rs index 0cd7056eea..758617175b 100644 --- a/src/enrichment_data/src/enrichment_table/mod.rs +++ b/src/enrichment_data/src/enrichment_table/mod.rs @@ -177,7 +177,7 @@ pub async fn save_enrichment_data( let db_schema = stream_schema_map .get(&stream_name) .map(|s| s.schema().as_ref().clone()) - .unwrap_or(Schema::empty()); + .unwrap_or_else(Schema::empty); if !db_schema.fields().is_empty() && db_schema.fields().ne(inferred_schema.fields()) { log::error!("Schema mismatch for enrichment table {org_id}/{stream_name}"); return Ok(MetaHttpResponse::error_with_header( diff --git a/src/enrichment_data/src/enrichment_table/url_processor.rs b/src/enrichment_data/src/enrichment_table/url_processor.rs index f08a4374dc..7b2b69152b 100644 --- a/src/enrichment_data/src/enrichment_table/url_processor.rs +++ b/src/enrichment_data/src/enrichment_table/url_processor.rs @@ -1665,7 +1665,7 @@ async fn process_enrichment_table_url( let schema = stream_schema_map .get(&stream_name) .map(|s| s.schema().as_ref().clone()) - .unwrap_or(arrow_schema::Schema::empty()); + .unwrap_or_else(arrow_schema::Schema::empty); // Calculate total expected size let current_size_in_bytes = if append_data { @@ -1844,7 +1844,7 @@ async fn save_enrichment_batch( let db_schema = stream_schema_map .get(&stream_name) .map(|s| s.schema().as_ref().clone()) - .unwrap_or(arrow_schema::Schema::empty()); + .unwrap_or_else(arrow_schema::Schema::empty); if !db_schema.fields().is_empty() && db_schema.fields().ne(inferred_schema.fields()) { return Err(anyhow!( diff --git a/src/infra/src/coordinator/alerts.rs b/src/infra/src/coordinator/alerts.rs index 2e2bf02150..f529cd8f42 100644 --- a/src/infra/src/coordinator/alerts.rs +++ b/src/infra/src/coordinator/alerts.rs @@ -35,7 +35,7 @@ pub async fn emit_put_event( cluster_coordinator .put( &key, - bytes::Bytes::from(folder_id.unwrap_or("default".to_string())), + bytes::Bytes::from(folder_id.unwrap_or_else(|| "default".to_string())), true, None, ) diff --git a/src/infra/src/table/ratelimit.rs b/src/infra/src/table/ratelimit.rs index c28e415198..f4396c8c7b 100644 --- a/src/infra/src/table/ratelimit.rs +++ b/src/infra/src/table/ratelimit.rs @@ -165,7 +165,7 @@ async fn add_batch(rules: Vec) -> Result<(), anyhow::Error> { rule_id: Set(rule.rule_id.unwrap()), rule_type: Set(rule .rule_type - .unwrap_or(RatelimitRuleType::Exact.to_string())), + .unwrap_or_else(|| RatelimitRuleType::Exact.to_string())), user_role: Set(rule.user_role.unwrap_or_default()), user_id: Set(rule.user_id.unwrap_or_default()), api_group_name: Set(rule.api_group_name.unwrap_or_default()), @@ -245,7 +245,7 @@ async fn add_upsert_batch(rules: Vec) -> Result<(), anyhow::Error active_model.rule_id = Set(existing.rule_id.clone()); active_model.rule_type = Set(rule .rule_type - .unwrap_or(RatelimitRuleType::Exact.to_string())); + .unwrap_or_else(|| RatelimitRuleType::Exact.to_string())); active_model.user_role = Set(rule.user_role.unwrap_or_default()); active_model.user_id = Set(rule.user_id.unwrap_or_default()); active_model.api_group_name = Set(rule.api_group_name.unwrap_or_default()); @@ -278,7 +278,7 @@ async fn add_upsert_batch(rules: Vec) -> Result<(), anyhow::Error rule_id: Set(rule.rule_id.unwrap()), rule_type: Set(rule .rule_type - .unwrap_or(RatelimitRuleType::Exact.to_string())), + .unwrap_or_else(|| RatelimitRuleType::Exact.to_string())), user_role: Set(rule.user_role.unwrap_or_default()), user_id: Set(rule.user_id.unwrap_or_default()), api_group_name: Set(rule.api_group_name.unwrap_or_default()), @@ -322,7 +322,7 @@ async fn add_single(rule: RatelimitRule) -> Result<(), anyhow::Error> { rule_id: Set(rule.rule_id.unwrap()), rule_type: Set(rule .rule_type - .unwrap_or(RatelimitRuleType::Exact.to_string())), + .unwrap_or_else(|| RatelimitRuleType::Exact.to_string())), user_role: Set(rule.user_role.unwrap_or("".to_string())), user_id: Set(rule.user_id.unwrap_or_default()), api_group_name: Set(rule.api_group_name.unwrap_or_default()), @@ -372,7 +372,7 @@ async fn update_single(rule: RatelimitRule) -> Result<(), anyhow::Error> { Column::RuleType, Expr::value( rule.rule_type - .unwrap_or(RatelimitRuleType::Exact.to_string()), + .unwrap_or_else(|| RatelimitRuleType::Exact.to_string()), ), ) .col_expr(Column::Threshold, Expr::value(rule.threshold)) @@ -460,7 +460,7 @@ async fn update_batch(rules: Vec) -> Result<(), anyhow::Error> { Column::RuleType, Expr::value( rule.rule_type - .unwrap_or(RatelimitRuleType::Exact.to_string()), + .unwrap_or_else(|| RatelimitRuleType::Exact.to_string()), ), ) .col_expr(Column::Threshold, Expr::value(rule.threshold)) diff --git a/src/metrics_index/src/pruner.rs b/src/metrics_index/src/pruner.rs index e818704c67..5b9b68417e 100644 --- a/src/metrics_index/src/pruner.rs +++ b/src/metrics_index/src/pruner.rs @@ -75,7 +75,9 @@ pub async fn search( let mut index_files = BTreeMap::new(); for file in files.iter() { // only indexed metrics files own a sidecar; other layouts stay as they are - if MetricsFileLayout::of(&file.key) != Some(MetricsFileLayout::Indexed) { + if index_files.contains_key(&file.key) + || MetricsFileLayout::of(&file.key) != Some(MetricsFileLayout::Indexed) + { continue; } let Some(sidecar_path) = MetricsFileLayout::metrics_index_path(&file.key) else { @@ -104,12 +106,10 @@ pub async fn search( } else { String::new() }; - index_files.entry(file.key.clone()).or_insert(( - file.account.clone(), - sidecar_path, - cache_key, - expected_rows, - )); + index_files.insert( + file.key.clone(), + (file.account.clone(), sidecar_path, cache_key, expected_rows), + ); } if index_files.is_empty() { return Ok(None); diff --git a/src/search/src/tantivy/mod.rs b/src/search/src/tantivy/mod.rs index 60ec3b9437..8ce0dbf866 100644 --- a/src/search/src/tantivy/mod.rs +++ b/src/search/src/tantivy/mod.rs @@ -470,7 +470,7 @@ async fn search_tantivy_index( // generate the tantivy query let condition: IndexCondition = - index_condition.ok_or(anyhow::anyhow!("IndexCondition not found"))?; + index_condition.ok_or_else(|| anyhow::anyhow!("IndexCondition not found"))?; let (mut query, has_skipped_conditions) = condition.to_tantivy_query(trace_id, tantivy_schema.clone(), fts_field)?; diff --git a/src/search_service/src/cache/mod.rs b/src/search_service/src/cache/mod.rs index 2c041ed487..039064d09d 100644 --- a/src/search_service/src/cache/mod.rs +++ b/src/search_service/src/cache/mod.rs @@ -1131,10 +1131,7 @@ pub fn apply_vrl_to_response( .as_array() .unwrap() .iter() - .filter_map(|v| { - (!v.is_null()) - .then_some(config::utils::flatten::flatten(v.clone()).unwrap()) - }) + .filter_map(|v| super::flatten_vrl_result(v.clone())) .collect() } else { let mut error = "".to_string(); @@ -1155,8 +1152,7 @@ pub fn apply_vrl_to_response( if let Some(e) = err { error = e; } - (!ret_val.is_null()) - .then_some(config::utils::flatten::flatten(ret_val).unwrap()) + super::flatten_vrl_result(ret_val) }) .collect(); if !error.is_empty() { @@ -1267,6 +1263,61 @@ pub async fn apply_regex_to_response( mod tests { use super::*; + #[test] + fn test_apply_vrl_to_response_preserves_hits_on_compile_failure() { + let mut response = config::meta::search::Response { + hits: vec![json::json!({"nested": {"value": 42}})], + ..Default::default() + }; + let expected = response.hits.clone(); + let hits = apply_vrl_to_response( + Some(". = [".to_string()), + &mut response, + "default", + "test", + "test", + ); + assert_eq!(hits, expected); + } + + #[test] + fn test_apply_vrl_to_response_filters_null_per_record() { + let mut response = config::meta::search::Response { + hits: vec![ + json::json!({"drop": true}), + json::json!({"drop": false, "nested": {"value": 42}}), + ], + ..Default::default() + }; + let hits = apply_vrl_to_response( + Some("if .drop == true { . = null } else { del(.drop) }; .".to_string()), + &mut response, + "default", + "test", + "test", + ); + assert_eq!(hits, vec![json::json!({"nested_value": 42})]); + } + + #[test] + fn test_apply_vrl_to_response_filters_null_array_items() { + let mut response = config::meta::search::Response { + hits: vec![json::json!({"input": true})], + ..Default::default() + }; + let hits = apply_vrl_to_response( + Some( + "#ResultArray#SkipVRL#\n. = [null, {\"nested\": {\"value\": 42}}, null]" + .to_string(), + ), + &mut response, + "default", + "test", + "test", + ); + assert_eq!(hits, vec![json::json!({"nested_value": 42})]); + } + #[test] fn test_is_result_array_skip_vrl() { let query_fn = r#"#ResultArray#SkipVRL# diff --git a/src/search_service/src/cluster/http.rs b/src/search_service/src/cluster/http.rs index 3aeff71101..62b8ff6b6f 100644 --- a/src/search_service/src/cluster/http.rs +++ b/src/search_service/src/cluster/http.rs @@ -144,7 +144,7 @@ pub async fn search( ret_val .as_array() - .ok_or(Error::Message("Expected array".to_string()))? + .ok_or_else(|| Error::Message("Expected array".to_string()))? .iter() .filter_map(|v| { if !v.is_null() && v.is_object() { diff --git a/src/search_service/src/file_list.rs b/src/search_service/src/file_list.rs index 855cc0da84..6774b5d020 100644 --- a/src/search_service/src/file_list.rs +++ b/src/search_service/src/file_list.rs @@ -180,7 +180,8 @@ pub async fn query_by_ids( db_files.extend( dumped_files .iter() - .filter_map(|file| ids_set.contains(&file.id).then_some(file.into())), + .filter(|file| ids_set.contains(&file.id)) + .map(|file| file.into()), ); } diff --git a/src/search_service/src/grpc/flight.rs b/src/search_service/src/grpc/flight.rs index 4cbe1b7df2..590e885561 100644 --- a/src/search_service/src/grpc/flight.rs +++ b/src/search_service/src/grpc/flight.rs @@ -150,7 +150,7 @@ pub async fn search( let db_schema = infra::schema::get(&org_id, &stream_name, stream_type) .await - .unwrap_or(arrow_schema::Schema::empty()); + .unwrap_or_else(|_| arrow_schema::Schema::empty()); let stream_settings = unwrap_stream_settings(&db_schema); let stream_created_at = unwrap_stream_created_at(&db_schema); let fst_fields = get_stream_setting_fts_fields(&stream_settings) @@ -178,11 +178,8 @@ pub async fn search( .index_info .equal_keys .iter() - .filter_map(|v| { - latest_schema_map - .contains_key(&v.key) - .then_some((v.key.to_string(), v.value.to_string())) - }) + .filter(|v| latest_schema_map.contains_key(&v.key)) + .map(|v| (v.key.to_string(), v.value.to_string())) .collect::>(); // get all tables diff --git a/src/search_service/src/lib.rs b/src/search_service/src/lib.rs index a0eb051f18..2b5202a713 100644 --- a/src/search_service/src/lib.rs +++ b/src/search_service/src/lib.rs @@ -511,8 +511,7 @@ pub async fn search_multi( .collect::>(); Some(serde_json::Value::Array(flattened_array)) } else { - (!v.is_null()) - .then_some(config::utils::flatten::flatten(v.clone()).unwrap()) + flatten_vrl_result(v.clone()) } }) .collect() @@ -535,8 +534,7 @@ pub async fn search_multi( if let Some(e) = err { error = e; } - (!ret_val.is_null()) - .then_some(config::utils::flatten::flatten(ret_val).unwrap()) + flatten_vrl_result(ret_val) }) .collect(); if !error.is_empty() { @@ -1089,6 +1087,10 @@ pub fn check_search_allowed(_org_id: &str, _stream: Option<&str>) -> Result<(), Ok(()) } +fn flatten_vrl_result(value: json::Value) -> Option { + (!value.is_null()).then(|| config::utils::flatten::flatten(value).unwrap()) +} + #[cfg(test)] mod tests { use std::sync::Arc; @@ -1098,6 +1100,27 @@ mod tests { use super::*; + #[test] + fn test_flatten_vrl_result_filters_null_and_preserves_nested_records() { + assert_eq!(flatten_vrl_result(json::Value::Null), None); + assert_eq!( + flatten_vrl_result(json::json!({"nested": {"value": 42}})), + Some(json::json!({"nested_value": 42})) + ); + let values = vec![ + json::Value::Null, + json::json!({"nested": {"value": 42}}), + json::Value::Null, + ]; + assert_eq!( + values + .into_iter() + .filter_map(flatten_vrl_result) + .collect::>(), + vec![json::json!({"nested_value": 42})] + ); + } + #[test] fn test_generate_filter_from_equal_items_empty() { let result = generate_filter_from_equal_items(&[]); diff --git a/src/search_service/src/partition/sql_context.rs b/src/search_service/src/partition/sql_context.rs index a21f7ccd8d..e899040a60 100644 --- a/src/search_service/src/partition/sql_context.rs +++ b/src/search_service/src/partition/sql_context.rs @@ -64,7 +64,8 @@ impl PartitionSqlContext { let ts_column = get_ts_col_order_by(&sql, TIMESTAMP_COL_NAME, is_complex).map(|(v, _)| v); let is_streaming_agg = is_streaming_aggregate(&req.sql, ts_column.as_deref()); let apply_over_hits = req.query_fn.as_ref().is_some_and(|v| { - !v.is_empty() && RESULT_ARRAY.is_match(&base64::decode_url(v).unwrap_or(v.to_string())) + !v.is_empty() + && RESULT_ARRAY.is_match(&base64::decode_url(v).unwrap_or_else(|_| v.to_string())) }); let use_single_partition = is_explain diff --git a/src/stream/src/lib.rs b/src/stream/src/lib.rs index cf8e41e9bf..33b1053d37 100644 --- a/src/stream/src/lib.rs +++ b/src/stream/src/lib.rs @@ -176,7 +176,7 @@ pub fn stream_res( stats.created_at = unwrap_stream_created_at(&schema).unwrap_or_default(); let metrics_meta = if stream_type == StreamType::Metrics { - let mut meta = get_prom_metadata_from_schema(&schema).unwrap_or(promql::Metadata { + let mut meta = get_prom_metadata_from_schema(&schema).unwrap_or_else(|| promql::Metadata { metric_type: promql::MetricType::Empty, metric_family_name: stream_name.to_string(), help: stream_name.to_string(), diff --git a/src/super_cluster_queue/src/pipelines.rs b/src/super_cluster_queue/src/pipelines.rs index 5c1b704776..169fa36368 100644 --- a/src/super_cluster_queue/src/pipelines.rs +++ b/src/super_cluster_queue/src/pipelines.rs @@ -25,7 +25,7 @@ pub(crate) async fn process(msg: Message) -> Result<()> { MessageType::PipelinePut => { let bytes = msg .value - .ok_or(Error::Message("Message missing value".to_string()))?; + .ok_or_else(|| Error::Message("Message missing value".to_string()))?; let pipeline: Pipeline = json::from_slice(&bytes).inspect_err(|e| { log::error!( "[SUPER_CLUSTER:PIPELINE] Failed to deserialize message value to pipeline: {e}" @@ -42,7 +42,7 @@ pub(crate) async fn process(msg: Message) -> Result<()> { MessageType::EnrichmentUrlPut => { let bytes = msg .value - .ok_or(Error::Message("Message missing value".to_string()))?; + .ok_or_else(|| Error::Message("Message missing value".to_string()))?; let mut enrichment_url_job: enrichment_table_urls::EnrichmentTableUrlRecord = json::from_slice(&bytes).inspect_err(|e| { log::error!( "[SUPER_CLUSTER:PIPELINE] Failed to deserialize message value to pipeline: {e}" @@ -64,7 +64,7 @@ pub(crate) async fn process(msg: Message) -> Result<()> { // format is /{BACKFILL_JOBS_KEY}/org/job_id let bytes = msg .value - .ok_or(Error::Message("Message missing value".to_string()))?; + .ok_or_else(|| Error::Message("Message missing value".to_string()))?; let backfill_job: backfill_jobs::BackfillJob = json::from_slice(&bytes).inspect_err(|e| { log::error!( "[SUPER_CLUSTER:BACKFILL] Failed to deserialize message value to backfill job: {e}" @@ -76,7 +76,7 @@ pub(crate) async fn process(msg: Message) -> Result<()> { // format is /{BACKFILL_JOBS_KEY}/org/job_id let bytes = msg .value - .ok_or(Error::Message("Message missing value".to_string()))?; + .ok_or_else(|| Error::Message("Message missing value".to_string()))?; let backfill_job: backfill_jobs::BackfillJob = json::from_slice(&bytes).inspect_err(|e| { log::error!( "[SUPER_CLUSTER:BACKFILL] Failed to deserialize message value to backfill job: {e}" diff --git a/src/super_cluster_queue/src/ratelimit.rs b/src/super_cluster_queue/src/ratelimit.rs index 4cd1d25615..5bd694b1ba 100644 --- a/src/super_cluster_queue/src/ratelimit.rs +++ b/src/super_cluster_queue/src/ratelimit.rs @@ -28,7 +28,7 @@ pub async fn process(msg: Message) -> Result<()> { let bytes = msg .value - .ok_or(Error::Message("Message missing value".to_string()))?; + .ok_or_else(|| Error::Message("Message missing value".to_string()))?; let rule = RuleEntry::try_from(&bytes).map_err(|e| Error::Message(e.to_string()))?; match msg.message_type { diff --git a/src/synthetics/src/status_pages.rs b/src/synthetics/src/status_pages.rs index 26fa92d3cb..a69f53833c 100644 --- a/src/synthetics/src/status_pages.rs +++ b/src/synthetics/src/status_pages.rs @@ -318,7 +318,7 @@ async fn apply_engine_actions( // Latest resolution wins for the merge window. let entry = recently_resolved .entry(check_id) - .or_insert((n.id.clone(), resolved_at)); + .or_insert_with(|| (n.id.clone(), resolved_at)); if resolved_at > entry.1 { *entry = (n.id.clone(), resolved_at); } diff --git a/src/usage_reporting/src/lib.rs b/src/usage_reporting/src/lib.rs index caec0f85fe..53d71761e6 100644 --- a/src/usage_reporting/src/lib.rs +++ b/src/usage_reporting/src/lib.rs @@ -242,7 +242,7 @@ pub async fn report_request_usage_stats( return; let now = DateTime::from_timestamp_micros(timestamp).unwrap(); - let request_body = stats.request_body.unwrap_or(usage_type.to_string()); + let request_body = stats.request_body.unwrap_or_else(|| usage_type.to_string()); let user_email = stats.user_email.unwrap_or_default(); let mut usages = Vec::with_capacity(if num_functions > 0 { 2 } else { 1 });