fix(profiles): route queries to queriers and reject bad profiles input (#16094)

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,

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,

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,

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<String, String>) -> 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<String, String>,
+) -> Result<ProfilesQueryBody, Response> {
+    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<String, String>) -> 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<T: std::str::FromStr>(
+    params: &HashMap<String, String>,
+    key: &str,
+) -> Result<Option<T>, 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<ProfileFilter> {
+fn parse_filters_param(raw: Option<&str>) -> Result<Vec<ProfileFilter>, 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<ProfilesQueryBody, Response> {
     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<String, String>,
-) {
+) -> 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<i64>) -> Result<i64, Response> {
@@ -411,6 +421,7 @@ pub(crate) fn resolve_timeout(timeout: Option<i64>) -> Result<i64, Response> {
 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(&params).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(&params).unwrap().max_nodes.is_none());
+    }
+
     #[test]
     fn tag_filter_uses_column_equality() {
         let body = ProfilesQueryBody {

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,

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(&params);
+    let mut body = match query_body_from_params(&params) {
+        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,

src/config/src/router.rs

@@ -14,7 +14,7 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 /// 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]

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<String, json::Value>,
+) -> Result<arrow_schema::Schema, anyhow::Error> {
+    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();

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)}