diff --git a/src/api/search/src/profiles/diff.rs b/src/api/search/src/profiles/diff.rs index d3dbea56de..22ee862390 100644 --- a/src/api/search/src/profiles/diff.rs +++ b/src/api/search/src/profiles/diff.rs @@ -160,6 +160,7 @@ pub async fn diff_profiles( let params_for = |trace_id, side: &DiffSide| ProfilesSqlParams { request_trace_id: trace_id, org_id: &org_id, + stream_name: &stream_name, user_id: &user_email.user_id, start_time: side.start_time, end_time: side.end_time, diff --git a/src/api/search/src/profiles/merge.rs b/src/api/search/src/profiles/merge.rs index c4878b039f..02de3bc5c0 100644 --- a/src/api/search/src/profiles/merge.rs +++ b/src/api/search/src/profiles/merge.rs @@ -208,6 +208,7 @@ async fn merge_profiles_impl( let sql_params = ProfilesSqlParams { request_trace_id: &request_trace_id, org_id: &org_id, + stream_name: &stream_name, user_id: &user_email.user_id, start_time, end_time, diff --git a/src/api/search/src/profiles/meta.rs b/src/api/search/src/profiles/meta.rs index 1546028034..fe821d1cc2 100644 --- a/src/api/search/src/profiles/meta.rs +++ b/src/api/search/src/profiles/meta.rs @@ -103,6 +103,7 @@ pub async fn get_profiles_meta( let sql_params = ProfilesSqlParams { request_trace_id: &request_trace_id, org_id: &org_id, + stream_name: &stream_name, user_id: &user_email.user_id, start_time, end_time, diff --git a/src/api/search/src/profiles/mod.rs b/src/api/search/src/profiles/mod.rs index 7196d9ca33..11dc1e3c90 100644 --- a/src/api/search/src/profiles/mod.rs +++ b/src/api/search/src/profiles/mod.rs @@ -47,7 +47,7 @@ use utoipa::ToSchema; use crate::{ common::meta::http::HttpResponse as MetaHttpResponse, - search::error_utils::map_error_to_http_response, + search::{error_utils::map_error_to_http_response, utils::enrich_field_error}, }; const DEFAULT_RESULT_SIZE: i64 = 99_999; @@ -167,16 +167,12 @@ pub(crate) fn parse_time_range_from_params( parse_required_time_range(start_time, end_time) } -pub(crate) fn query_body_from_params(params: &HashMap) -> ProfilesQueryBody { - ProfilesQueryBody { - start_time: params - .get("start_time") - .and_then(|v| v.parse().ok()) - .unwrap_or_default(), - end_time: params - .get("end_time") - .and_then(|v| v.parse().ok()) - .unwrap_or_default(), +pub(crate) fn query_body_from_params( + params: &HashMap, +) -> Result { + Ok(ProfilesQueryBody { + start_time: parse_param(params, "start_time")?.unwrap_or_default(), + end_time: parse_param(params, "end_time")?.unwrap_or_default(), data_source: params.get("data_source").cloned().filter(|s| !s.is_empty()), service_name: params .get("service_name") @@ -190,29 +186,42 @@ pub(crate) fn query_body_from_params(params: &HashMap) -> Profil .get("profile_unit") .cloned() .filter(|s| !s.is_empty()), - filters: parse_filters_param(params.get("filters").map(|s| s.as_str())), + filters: parse_filters_param(params.get("filters").map(|s| s.as_str()))?, step: params.get("step").cloned().filter(|s| !s.is_empty()), - max_nodes: params.get("max_nodes").and_then(|v| v.parse().ok()), - timeout: params.get("timeout").and_then(|v| v.parse().ok()), + max_nodes: parse_param(params, "max_nodes")?, + timeout: parse_param(params, "timeout")?, group_by: params.get("group_by").cloned().filter(|s| !s.is_empty()), - group_limit: params.get("group_limit").and_then(|v| v.parse().ok()), + group_limit: parse_param(params, "group_limit")?, + }) +} + +/// An empty value counts as absent; anything else must parse. +fn parse_param( + params: &HashMap, + key: &str, +) -> Result, Response> { + match params.get(key).filter(|v| !v.is_empty()) { + None => Ok(None), + Some(v) => v + .parse() + .map(Some) + .map_err(|_| MetaHttpResponse::bad_request(format!("Invalid {key} parameter"))), } } /// `filters` query param: `key=value,key2=value2`. Each side is percent-encoded. -fn parse_filters_param(raw: Option<&str>) -> Vec { +fn parse_filters_param(raw: Option<&str>) -> Result, Response> { let Some(raw) = raw.filter(|s| !s.is_empty()) else { - return Vec::new(); + return Ok(Vec::new()); }; raw.split(',') - .filter_map(|pair| { - let (key, value) = pair.split_once('=')?; - let key = decode_filter_component(key); - let value = decode_filter_component(value); - if key.is_empty() { - return None; - } - Some(ProfileFilter { + .map(|pair| { + let (key, value) = pair + .split_once('=') + .map(|(k, v)| (decode_filter_component(k), decode_filter_component(v))) + .filter(|(k, _)| !k.is_empty()) + .ok_or_else(|| MetaHttpResponse::bad_request(format!("invalid filter: {pair}")))?; + Ok(ProfileFilter { key, op: "=".to_string(), value, @@ -334,13 +343,13 @@ pub(crate) fn resolve_query_body( ) -> Result { match body { Ok(Json(mut body)) => { - merge_params_into_body(&mut body, params); + merge_params_into_body(&mut body, params)?; Ok(body) } Err(rejection) => { // GET (and HEAD) may omit a JSON body; fall back to query params. if matches!(*method, Method::GET | Method::HEAD) { - Ok(query_body_from_params(params)) + query_body_from_params(params) } else { Err(MetaHttpResponse::bad_request(rejection.body_text())) } @@ -351,14 +360,14 @@ pub(crate) fn resolve_query_body( pub(crate) fn merge_params_into_body( body: &mut ProfilesQueryBody, params: &HashMap, -) { +) -> Result<(), Response> { if body.start_time == 0 - && let Some(v) = params.get("start_time").and_then(|s| s.parse().ok()) + && let Some(v) = parse_param(params, "start_time")? { body.start_time = v; } if body.end_time == 0 - && let Some(v) = params.get("end_time").and_then(|s| s.parse().ok()) + && let Some(v) = parse_param(params, "end_time")? { body.end_time = v; } @@ -387,17 +396,18 @@ pub(crate) fn merge_params_into_body( body.step = params.get("step").cloned().filter(|s| !s.is_empty()); } if body.max_nodes.is_none() { - body.max_nodes = params.get("max_nodes").and_then(|s| s.parse().ok()); + body.max_nodes = parse_param(params, "max_nodes")?; } if body.filters.is_empty() { - body.filters = parse_filters_param(params.get("filters").map(|s| s.as_str())); + body.filters = parse_filters_param(params.get("filters").map(|s| s.as_str()))?; } if body.group_by.is_none() { body.group_by = params.get("group_by").cloned().filter(|s| !s.is_empty()); } if body.group_limit.is_none() { - body.group_limit = params.get("group_limit").and_then(|s| s.parse().ok()); + body.group_limit = parse_param(params, "group_limit")?; } + Ok(()) } pub(crate) fn resolve_timeout(timeout: Option) -> Result { @@ -411,6 +421,7 @@ pub(crate) fn resolve_timeout(timeout: Option) -> Result { pub(crate) struct ProfilesSqlParams<'a> { pub request_trace_id: &'a str, pub org_id: &'a str, + pub stream_name: &'a str, pub user_id: &'a str, pub start_time: i64, pub end_time: i64, @@ -469,10 +480,15 @@ pub(crate) async fn run_profiles_sql( .await { Ok(response) => reject_partial_search_response(response), - Err(error) => Err(map_error_to_http_response( - &error, - Some(params.request_trace_id.to_string()), - )), + Err(error) => { + let streams = [params.stream_name.to_string()]; + let error = + enrich_field_error(error, params.org_id, StreamType::Profiles, &streams).await; + Err(map_error_to_http_response( + &error, + Some(params.request_trace_id.to_string()), + )) + } } } @@ -534,7 +550,7 @@ mod tests { #[test] fn parse_filters_param_keeps_a_comma_inside_a_value() { - let filters = parse_filters_param(Some("k8s_pod_name=a%2Cb,env=prod")); + let filters = parse_filters_param(Some("k8s_pod_name=a%2Cb,env=prod")).unwrap(); assert_eq!(filters.len(), 2); assert_eq!(filters[0].key, "k8s_pod_name"); assert_eq!(filters[0].value, "a,b"); @@ -544,11 +560,42 @@ mod tests { #[test] fn parse_filters_param_reads_plain_pairs() { - let filters = parse_filters_param(Some("env=prod,region=us")); + let filters = parse_filters_param(Some("env=prod,region=us")).unwrap(); assert_eq!(filters[0].value, "prod"); assert_eq!(filters[1].value, "us"); } + async fn bad_request_text(resp: Response) -> String { + assert_eq!(resp.status(), axum::http::StatusCode::BAD_REQUEST); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + String::from_utf8(bytes.to_vec()).unwrap() + } + + #[tokio::test] + async fn parse_filters_param_rejects_a_malformed_pair() { + for raw in ["foo", "env=prod,foo", "=prod"] { + let resp = parse_filters_param(Some(raw)).unwrap_err(); + assert!( + bad_request_text(resp).await.contains("invalid filter: "), + "{raw}" + ); + } + } + + #[tokio::test] + async fn get_params_reject_unparsable_numbers() { + for key in ["max_nodes", "timeout", "group_limit", "start_time"] { + let params = HashMap::from([(key.to_string(), "abc".to_string())]); + let resp = query_body_from_params(¶ms).unwrap_err(); + let text = bad_request_text(resp).await; + assert!(text.contains(&format!("Invalid {key} parameter")), "{text}"); + } + let params = HashMap::from([("max_nodes".to_string(), String::new())]); + assert!(query_body_from_params(¶ms).unwrap().max_nodes.is_none()); + } + #[test] fn tag_filter_uses_column_equality() { let body = ProfilesQueryBody { diff --git a/src/api/search/src/profiles/series.rs b/src/api/search/src/profiles/series.rs index 1a9c4ce7bf..410cfc780a 100644 --- a/src/api/search/src/profiles/series.rs +++ b/src/api/search/src/profiles/series.rs @@ -280,6 +280,7 @@ async fn profiles_series_impl( let sql_params = ProfilesSqlParams { request_trace_id: &request_trace_id, org_id: &org_id, + stream_name: &stream_name, user_id: &user_email.user_id, start_time, end_time, diff --git a/src/api/search/src/profiles/tag_values.rs b/src/api/search/src/profiles/tag_values.rs index 3d239d77d4..64ffb0b101 100644 --- a/src/api/search/src/profiles/tag_values.rs +++ b/src/api/search/src/profiles/tag_values.rs @@ -113,7 +113,10 @@ pub async fn get_profiles_tag_values( Ok(range) => range, Err(response) => return response, }; - let mut body = query_body_from_params(¶ms); + let mut body = match query_body_from_params(¶ms) { + Ok(body) => body, + Err(response) => return response, + }; body.start_time = start_time; body.end_time = end_time; // Drop filters that target the tag whose values we are listing. @@ -136,6 +139,7 @@ pub async fn get_profiles_tag_values( let sql_params = ProfilesSqlParams { request_trace_id: &request_trace_id, org_id: &org_id, + stream_name: &stream_name, user_id: &user_email.user_id, start_time, end_time, diff --git a/src/config/src/router.rs b/src/config/src/router.rs index bbe058c4f8..884ce404d8 100644 --- a/src/config/src/router.rs +++ b/src/config/src/router.rs @@ -14,7 +14,7 @@ // along with this program. If not, see . /// usize indicates the number of parts to skip based on their actual paths. -const QUERIER_ROUTES: [(&str, usize); 38] = [ +const QUERIER_ROUTES: [(&str, usize); 43] = [ ("config", 0), // /config (unauthenticated bootstrap) ("config", 2), // /api/{org_id}/config (authenticated full) ("summary", 2), // /api/{org_id}/summary @@ -30,6 +30,11 @@ const QUERIER_ROUTES: [(&str, usize); 38] = [ ("details", 5), // /api/{org_id}/{stream_name}/traces/{trace_id}/details ("traces/time_range", 3), // /api/{org_id}/{stream_name}/traces/time_range ("traces/time_range", 2), // /api/{org_id}/traces/time_range + ("profiles/meta", 3), // /api/{org_id}/{stream_name}/profiles/meta + ("profiles/tag_values", 3), // /api/{org_id}/{stream_name}/profiles/tag_values + ("profiles/series", 3), // /api/{org_id}/{stream_name}/profiles/series + ("profiles/merge", 3), // /api/{org_id}/{stream_name}/profiles/merge + ("profiles/diff", 3), // /api/{org_id}/{stream_name}/profiles/diff ("rum/lookup", 2), // /api/{org_id}/rum/lookup ("clusters", 1), // /api/clusters ("query_manager", 2), // /api/{org_id}/query_manager/... @@ -237,6 +242,12 @@ mod tests { )); // the org-level entry must not swallow the OTLP ingest route assert!(!is_querier_route("/api/org1/traces")); + for leaf in ["meta", "tag_values", "series", "merge", "diff"] { + assert!(is_querier_route(&format!( + "/api/org1/default/profiles/{leaf}" + ))); + } + assert!(!is_querier_route("/api/org1/v1/profiles")); } #[test] diff --git a/src/core/src/profiles/mod.rs b/src/core/src/profiles/mod.rs index 41a34dbc28..7acb155694 100644 --- a/src/core/src/profiles/mod.rs +++ b/src/core/src/profiles/mod.rs @@ -415,11 +415,7 @@ async fn buffer_records( if !stream_schema_map.contains_key(stream_name) { let mut schema = infra::schema::get(org_id, stream_name, StreamType::Profiles).await?; if schema.fields().is_empty() { - let first_str = json::to_string(&records[0]).unwrap(); - let mut schema_reader = BufReader::new(first_str.as_bytes()); - let inferred_schema = - infer_json_schema(&mut schema_reader, None, StreamType::Profiles)?; - schema = inferred_schema; + schema = new_stream_schema(org_id, stream_name, &records[0])?; db::schema::merge( org_id, stream_name, @@ -485,6 +481,25 @@ async fn buffer_records( Ok(()) } +/// check_for_schema takes its fast path once this schema is saved, so the columns limit is checked +/// here. +fn new_stream_schema( + org_id: &str, + stream_name: &str, + first_record: &json::Map, +) -> Result { + let first_str = json::to_string(first_record).unwrap(); + let mut schema_reader = BufReader::new(first_str.as_bytes()); + let schema = infer_json_schema(&mut schema_reader, None, StreamType::Profiles)?; + schema::check_request_columns_limit( + org_id, + StreamType::Profiles, + stream_name, + schema.fields().len(), + )?; + Ok(schema) +} + /// Redacts flattened profile records before they are buffered for the WAL. #[cfg(feature = "vectorscan")] async fn apply_redaction( @@ -2031,6 +2046,21 @@ mod tests { assert_eq!(rejected, bounded); } + #[test] + fn new_stream_over_columns_limit_is_rejected() { + let limit = get_config().limit.req_cols_per_record_limit; + let mut record = json::Map::new(); + record.insert(TIMESTAMP_COL_NAME.to_string(), json::json!(now_micros())); + for i in 0..=limit { + record.insert(format!("attr_{i}"), json::json!("v")); + } + let err = new_stream_schema("test_org_id", "profiles_columns_limit", &record).unwrap_err(); + assert!( + err.to_string().contains("ZO_COLS_PER_RECORD_LIMIT"), + "{err}" + ); + } + #[test] fn collector_empty_attribute_wire_sentinel_survives_record_extraction() { let mut dictionary = ProfilesDictionary::decode(&[0x32, 0x02, 0x12, 0x00][..]).unwrap(); diff --git a/tests/api-testing/tests/profiles/test_profiles_ingest.py b/tests/api-testing/tests/profiles/test_profiles_ingest.py index 6b53488f2f..a9b0ff4661 100644 --- a/tests/api-testing/tests/profiles/test_profiles_ingest.py +++ b/tests/api-testing/tests/profiles/test_profiles_ingest.py @@ -517,8 +517,6 @@ def test_ing44_negative_value_kept(client: OpenObserveClient, prof_stream: str): assert wait_rows(client, prof_stream, 1)[0]["value"] == -5 -# Product bug: profiles never apply ZO_COLS_PER_RECORD_LIMIT; strict xfail turns red once o2-enterprise#3123 fixes it, then drop the marker. -@pytest.mark.xfail(strict=True, reason="Known product bug o2-enterprise#3123: profiles ignore ZO_COLS_PER_RECORD_LIMIT") def test_ing45_too_many_columns_rejected(client: OpenObserveClient, prof_stream: str): b = ProfilesBuilder() attrs = {f"extra.attr_{i:04d}": f"v{i}" for i in range(1010)}