From f8d8fe1071cf82dd2ac15872b3a067a9c63c1616 Mon Sep 17 00:00:00 2001 From: fengyue-xve <1759519137@qq.com> Date: Thu, 12 Nov 2026 01:11:21 +0000 Subject: [PATCH] fix: honor the pipeline version header and URL parameter in Elasticsearch _bulk (#9452) * fix: honor the pipeline version header and URL parameter in Elasticsearch _bulk The Elasticsearch _bulk handlers read only pipeline_name and msg_field and hardcoded the pipeline version to None, so the x-greptime-pipeline-version header and the version URL parameter were silently ignored and the latest version was always used. Resolve the version from the version query parameter, falling back to the x-greptime-pipeline-version header, and pass it to PipelineDefinition::from_name; an unparsable version now fails the request instead of being ignored. Without either source the latest version is used, as before. Signed-off-by: fengyue-xve <1759519137@qq.com> * fix(elasticsearch): reject invalid pipeline version headers and use HTTP item status Signed-off-by: fengyue-xve <1759519137@qq.com> --------- Signed-off-by: fengyue-xve <1759519137@qq.com> --- src/servers/src/elasticsearch.rs | 49 ++++++-- src/servers/src/http/extractor.rs | 2 +- tests-integration/tests/http.rs | 197 ++++++++++++++++++++++++++++++ 3 files changed, 240 insertions(+), 8 deletions(-) diff --git a/src/servers/src/elasticsearch.rs b/src/servers/src/elasticsearch.rs index 5ccda5ed7f4..f4788100e38 100644 --- a/src/servers/src/elasticsearch.rs +++ b/src/servers/src/elasticsearch.rs @@ -25,6 +25,7 @@ use common_error::ext::ErrorExt; use common_telemetry::{debug, error}; use headers::ContentType; use once_cell::sync::Lazy; +use pipeline::util::to_pipeline_version; use pipeline::{ GREPTIME_INTERNAL_IDENTITY_PIPELINE_NAME, GreptimePipelineParams, PipelineDefinition, }; @@ -44,7 +45,10 @@ use crate::http::event::{ LogIngesterQueryParams, LogState, PipelineIngestRequest, extract_pipeline_params_map_from_headers, ingest_logs_inner, }; -use crate::http::header::constants::GREPTIME_PIPELINE_NAME_HEADER_NAME; +use crate::http::extractor::string_value_from_header; +use crate::http::header::constants::{ + GREPTIME_PIPELINE_NAME_HEADER_NAME, GREPTIME_PIPELINE_VERSION_HEADER_NAME, +}; use crate::metrics::{ METRIC_ELASTICSEARCH_LOGS_DOCS_COUNT, METRIC_ELASTICSEARCH_LOGS_INGESTION_ELAPSED, }; @@ -155,6 +159,32 @@ async fn do_handle_bulk_api( .unwrap_or(GREPTIME_INTERNAL_IDENTITY_PIPELINE_NAME) }); + // The pipeline version can be selected with the `version` query parameter or the + // `x-greptime-pipeline-version` header; the query parameter takes precedence, consistent + // with how `pipeline_name` is resolved above. Without either, the latest version is used. + // Read the header through the shared extractor so a present but non-UTF-8 value is + // rejected with 400 instead of silently falling back to the latest pipeline. + let header_version = if params.version.is_none() { + match string_value_from_header(&headers, &[GREPTIME_PIPELINE_VERSION_HEADER_NAME]) { + Ok(version) => version, + Err((status, reason)) => { + return ( + status, + elasticsearch_headers(), + axum::Json(write_bulk_response( + start.elapsed().as_millis() as i64, + 0, + status.as_u16() as u32, + reason.as_str(), + )), + ); + } + } + } else { + None + }; + let pipeline_version = params.version.as_deref().or(header_version.as_deref()); + // Read the ndjson payload and convert it to a vector of Value. let requests = match parse_bulk_request(&payload, &index, ¶ms.msg_field) { Ok(requests) => requests, @@ -173,18 +203,22 @@ async fn do_handle_bulk_api( }; let log_num = requests.len(); - let pipeline = match PipelineDefinition::from_name(pipeline_name, None, None) { + let pipeline_result = to_pipeline_version(pipeline_version) + .and_then(|version| PipelineDefinition::from_name(pipeline_name, version, None)); + let pipeline = match pipeline_result { Ok(pipeline) => pipeline, Err(e) => { - // should be unreachable error!(e; "Failed to ingest logs"); + // Elasticsearch clients inspect `items[].create.status`, so report the + // HTTP status rather than GreptimeDB's internal status code. + let http_status = status_code_to_http_status(&e.status_code()); return ( - status_code_to_http_status(&e.status_code()), + http_status, elasticsearch_headers(), axum::Json(write_bulk_response( start.elapsed().as_millis() as i64, 0, - e.status_code() as u32, + http_status.as_u16() as u32, e.to_string().as_str(), )), ); @@ -202,13 +236,14 @@ async fn do_handle_bulk_api( .await { error!(e; "Failed to ingest logs"); + let http_status = status_code_to_http_status(&e.status_code()); return ( - status_code_to_http_status(&e.status_code()), + http_status, elasticsearch_headers(), axum::Json(write_bulk_response( start.elapsed().as_millis() as i64, 0, - e.status_code() as u32, + http_status.as_u16() as u32, e.to_string().as_str(), )), ); diff --git a/src/servers/src/http/extractor.rs b/src/servers/src/http/extractor.rs index e129ac6c1d9..75ebcb45743 100644 --- a/src/servers/src/http/extractor.rs +++ b/src/servers/src/http/extractor.rs @@ -220,7 +220,7 @@ fn parse_otlp_metric_translation_strategy( } #[inline] -fn string_value_from_header( +pub(crate) fn string_value_from_header( headers: &HeaderMap, header_keys: &[&str], ) -> Result, (StatusCode, String)> { diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index a7879d3a779..950feb8a6ca 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -227,6 +227,7 @@ macro_rules! http_tests { test_loki_json_logs_with_pipeline, test_elasticsearch_logs, test_elasticsearch_logs_with_index, + test_elasticsearch_pipeline_version, test_splunk_logs, test_splunk_raw, test_log_query, @@ -12191,6 +12192,202 @@ pub async fn test_elasticsearch_logs_with_index(store_type: StorageType) { guard.remove_all().await; } +pub async fn test_elasticsearch_pipeline_version(store_type: StorageType) { + common_telemetry::init_default_ut_logging(); + let (app, mut guard) = + setup_test_http_app_with_frontend(store_type, "test_elasticsearch_pipeline_version").await; + + let client = TestClient::new(app).await; + + // Create version 1 of the pipeline: it renames `name` to `name_v1`. + let pipeline_v1 = r#" +processors: + - date: + field: ts + formats: + - "%Y-%m-%dT%H:%M:%S%.fZ" +transform: + - field: name, name_v1 + type: string + - field: ts + type: time + index: timestamp +"#; + let res = client + .post("/v1/pipelines/esv") + .header("Content-Type", "application/x-yaml") + .body(pipeline_v1) + .send() + .await; + assert_eq!(res.status(), StatusCode::OK); + let content: Value = serde_json::from_str(&res.text().await).unwrap(); + let version_v1 = content + .get("pipelines") + .unwrap() + .as_array() + .unwrap() + .first() + .unwrap() + .get("version") + .unwrap() + .as_str() + .unwrap() + .to_string(); + + // Create version 2: it renames `name` to `name_v2` instead. + let res = client + .post("/v1/pipelines/esv") + .header("Content-Type", "application/x-yaml") + .body(pipeline_v1.replace("name_v1", "name_v2")) + .send() + .await; + assert_eq!(res.status(), StatusCode::OK); + let content: Value = serde_json::from_str(&res.text().await).unwrap(); + let version_v2 = content + .get("pipelines") + .unwrap() + .as_array() + .unwrap() + .first() + .unwrap() + .get("version") + .unwrap() + .as_str() + .unwrap() + .to_string(); + + let bulk_body = |index: &str| { + format!( + "{{\"create\":{{\"_index\":\"{index}\"}}}}\n{{\"name\":\"alice\",\"ts\":\"2024-01-01T00:00:00.000Z\"}}\n" + ) + }; + + // 1. Select version 1 with the `x-greptime-pipeline-version` header. + let res = send_req( + &client, + vec![ + ( + HeaderName::from_static("content-type"), + HeaderValue::from_static("application/json"), + ), + ( + HeaderName::from_static("x-greptime-pipeline-version"), + HeaderValue::from_str(&version_v1).unwrap(), + ), + ], + "/v1/elasticsearch/_bulk?pipeline_name=esv", + bulk_body("es_hdr").into_bytes(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + + validate_data( + "es_pipeline_version_header", + &client, + "select name_v1 from es_hdr;", + "[[\"alice\"]]", + ) + .await; + + // 2. Select version 1 with the `version` query parameter. + let encoded_version: String = + url::form_urlencoded::byte_serialize(version_v1.as_bytes()).collect(); + let res = send_req( + &client, + vec![( + HeaderName::from_static("content-type"), + HeaderValue::from_static("application/json"), + )], + &format!("/v1/elasticsearch/_bulk?pipeline_name=esv&version={encoded_version}"), + bulk_body("es_url").into_bytes(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + + validate_data( + "es_pipeline_version_url_param", + &client, + "select name_v1 from es_url;", + "[[\"alice\"]]", + ) + .await; + + // 3. Without a version, the latest version (2) is used. + let res = send_req( + &client, + vec![( + HeaderName::from_static("content-type"), + HeaderValue::from_static("application/json"), + )], + "/v1/elasticsearch/_bulk?pipeline_name=esv", + bulk_body("es_latest").into_bytes(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + + validate_data( + "es_pipeline_version_latest", + &client, + "select name_v2 from es_latest;", + "[[\"alice\"]]", + ) + .await; + + // 4. An invalid version is rejected with HTTP 400, and the bulk item carries the + // same HTTP status so Elasticsearch clients do not see an internal status code. + let res = send_req( + &client, + vec![( + HeaderName::from_static("content-type"), + HeaderValue::from_static("application/json"), + )], + "/v1/elasticsearch/_bulk?pipeline_name=esv&version=not-a-timestamp", + bulk_body("es_invalid_version").into_bytes(), + false, + ) + .await; + assert_eq!(StatusCode::BAD_REQUEST, res.status()); + let content: Value = serde_json::from_str(&res.text().await).unwrap(); + assert_eq!(content["errors"], json!(true)); + assert_eq!(content["items"][0]["create"]["status"], json!(400)); + + // 5. The query parameter takes precedence over a conflicting header: the header + // selects version 1 while the URL selects version 2, so `name_v2` must be written. + let encoded_version_v2: String = + url::form_urlencoded::byte_serialize(version_v2.as_bytes()).collect(); + let res = send_req( + &client, + vec![ + ( + HeaderName::from_static("content-type"), + HeaderValue::from_static("application/json"), + ), + ( + HeaderName::from_static("x-greptime-pipeline-version"), + HeaderValue::from_str(&version_v1).unwrap(), + ), + ], + &format!("/v1/elasticsearch/_bulk?pipeline_name=esv&version={encoded_version_v2}"), + bulk_body("es_precedence").into_bytes(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + + validate_data( + "es_pipeline_version_precedence", + &client, + "select name_v2 from es_precedence;", + "[[\"alice\"]]", + ) + .await; + + guard.remove_all().await; +} + pub async fn test_log_query(store_type: StorageType) { common_telemetry::init_default_ut_logging(); let (app, mut guard) = setup_test_http_app_with_frontend(store_type, "test_log_query").await;