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>
2026-11-14 06:46:56 +00:00 · 2026-11-12 01:11:21 +00:00
parent 6dbff8a081
commit f8d8fe1071
3 changed files with 240 additions and 8 deletions
@@ -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, &params.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(),
            )),
        );
@@ -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<Option<String>, (StatusCode, String)> {
@@ -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;