From ce428bb74dd54fbed374cbbe137298c4366fc94d Mon Sep 17 00:00:00 2001 From: YingNing <59612604+dxqlovestudy@users.noreply.github.com> Date: Fri, 16 Oct 2026 02:36:47 +0000 Subject: [PATCH] feat(servers): accept application/x-json-stream for log ingestion (#9581) * feat(servers): accept application/x-json-stream for log ingestion Support fluentbit's json_stream output format on the HTTP log ingestion endpoints by accepting `application/x-json-stream` as a payload content type. The payload is parsed as concatenated JSON -- multiple JSON values placed back-to-back without a separator, e.g. `{"a":1}{"b":2}` -- reusing the streaming value iterator already used for `application/json`, so the behavior for concatenated or mixed payloads is identical. Aimed at #5404 (msgpack support is left as a possible follow-up). Signed-off-by: dxqlovestudy <59612604+dxqlovestudy@users.noreply.github.com> * chore(servers): remove redundant comments per review feedback Signed-off-by: dxqlovestudy <59612604+dxqlovestudy@users.noreply.github.com> --------- Signed-off-by: dxqlovestudy <59612604+dxqlovestudy@users.noreply.github.com> --- src/servers/src/http/event.rs | 66 ++++++++++++++++++++++++++++++--- src/servers/src/http/header.rs | 3 ++ tests-integration/tests/http.rs | 56 +++++++++++++++++++++++++++- 3 files changed, 119 insertions(+), 6 deletions(-) diff --git a/src/servers/src/http/event.rs b/src/servers/src/http/event.rs index 0c96becaf59..651621dcdea 100644 --- a/src/servers/src/http/event.rs +++ b/src/servers/src/http/event.rs @@ -59,7 +59,8 @@ use crate::http::header::constants::{ GREPTIME_PIPELINE_PARAMS_HEADER, }; use crate::http::header::{ - CONTENT_TYPE_NDJSON_STR, CONTENT_TYPE_NDJSON_SUBTYPE_STR, CONTENT_TYPE_PROTOBUF_STR, + CONTENT_TYPE_JSON_STREAM_STR, CONTENT_TYPE_JSON_STREAM_SUBTYPE_STR, CONTENT_TYPE_NDJSON_STR, + CONTENT_TYPE_NDJSON_SUBTYPE_STR, CONTENT_TYPE_PROTOBUF_STR, }; use crate::http::result::greptime_manage_resp::{GreptimedbManageResponse, SqlOutput}; use crate::http::result::greptime_result_v1::GreptimedbV1Response; @@ -87,6 +88,8 @@ lazy_static! { ContentType::from_str(CONTENT_TYPE_PROTOBUF_STR).unwrap(); pub static ref NDJSON_CONTENT_TYPE: ContentType = ContentType::from_str(CONTENT_TYPE_NDJSON_STR).unwrap(); + pub static ref JSON_STREAM_CONTENT_TYPE: ContentType = + ContentType::from_str(CONTENT_TYPE_JSON_STREAM_STR).unwrap(); } /// LogIngesterQueryParams is used for query params of log ingester API. @@ -797,6 +800,10 @@ pub async fn log_ingester( #[derive(Debug, EnumIter)] enum EventPayloadResolverInner { Json, + /// `application/x-json-stream`: concatenated JSON values without any + /// separator between them, e.g. `{"a":1}{"b":2}` (the `json_stream` + /// format of fluentbit). + JsonStream, Ndjson, Text, } @@ -805,6 +812,7 @@ impl Display for EventPayloadResolverInner { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { EventPayloadResolverInner::Json => write!(f, "{}", *JSON_CONTENT_TYPE), + EventPayloadResolverInner::JsonStream => write!(f, "{}", *JSON_STREAM_CONTENT_TYPE), EventPayloadResolverInner::Ndjson => write!(f, "{}", *NDJSON_CONTENT_TYPE), EventPayloadResolverInner::Text => write!(f, "{}", *TEXT_CONTENT_TYPE), } @@ -818,6 +826,9 @@ impl TryFrom<&ContentType> for EventPayloadResolverInner { let mime: mime_guess::Mime = content_type.clone().into(); match (mime.type_(), mime.subtype()) { (mime::APPLICATION, mime::JSON) => Ok(EventPayloadResolverInner::Json), + (mime::APPLICATION, subtype) if subtype == CONTENT_TYPE_JSON_STREAM_SUBTYPE_STR => { + Ok(EventPayloadResolverInner::JsonStream) + } (mime::APPLICATION, subtype) if subtype == CONTENT_TYPE_NDJSON_SUBTYPE_STR => { Ok(EventPayloadResolverInner::Ndjson) } @@ -866,10 +877,12 @@ impl<'a> TryFrom<&'a ContentType> for EventPayloadResolver<'a> { impl EventPayloadResolver<'_> { fn parse_payload(&self, payload: Bytes, ignore_errors: bool) -> Result> { match self.inner { - EventPayloadResolverInner::Json => transform_ndjson_array_factory( - Deserializer::from_slice(&payload).into_iter(), - ignore_errors, - ), + EventPayloadResolverInner::Json | EventPayloadResolverInner::JsonStream => { + transform_ndjson_array_factory( + Deserializer::from_slice(&payload).into_iter(), + ignore_errors, + ) + } EventPayloadResolverInner::Ndjson => { let mut result = Vec::with_capacity(1000); let mut buffer = Buffers::new(1000); @@ -1077,6 +1090,49 @@ mod tests { assert_eq!(a, "[{\"a\":1},{\"b\":2}]"); } + #[test] + fn test_extract_by_json_stream_content_type() { + // `application/x-json-stream` accepts concatenated JSON values, with + // or without line breaks between them. + let json_stream = ContentType::from_str("application/x-json-stream").unwrap(); + let payload = r#"{"a": 1}{"b": 2} +{"c": 1}"# + .as_bytes(); + let payload = Bytes::from_static(payload); + + let values = + extract_pipeline_value_by_content_type(json_stream.clone(), payload, false).unwrap(); + assert_eq!( + values, + vec![ + json!({"a": 1}).into(), + json!({"b": 2}).into(), + json!({"c": 1}).into() + ] + ); + + assert!( + EventPayloadResolver::support_content_type_list().contains(&json_stream.to_string()) + ); + } + + #[test] + fn test_json_stream_matches_json_for_concatenated_values() { + // a concatenated payload is parsed exactly like `application/json` + let payload = Bytes::from_static(r#"{"a": 1}{"b": 2}"#.as_bytes()); + let via_json = + extract_pipeline_value_by_content_type(ContentType::json(), payload.clone(), false) + .unwrap(); + let via_stream = extract_pipeline_value_by_content_type( + ContentType::from_str("application/x-json-stream").unwrap(), + payload, + false, + ) + .unwrap(); + assert_eq!(via_json, via_stream); + assert_eq!(via_stream.len(), 2); + } + #[test] fn test_extract_by_content() { let payload = r#" diff --git a/src/servers/src/http/header.rs b/src/servers/src/http/header.rs index 625bee41ee7..d6f69fa72d5 100644 --- a/src/servers/src/http/header.rs +++ b/src/servers/src/http/header.rs @@ -106,6 +106,9 @@ pub static CONTENT_ENCODING_SNAPPY: HeaderValue = HeaderValue::from_static("snap pub static CONTENT_TYPE_NDJSON_STR: &str = "application/x-ndjson"; pub static CONTENT_TYPE_NDJSON_SUBTYPE_STR: &str = "x-ndjson"; +pub static CONTENT_TYPE_JSON_STREAM_STR: &str = "application/x-json-stream"; +pub static CONTENT_TYPE_JSON_STREAM_SUBTYPE_STR: &str = "x-json-stream"; + pub struct GreptimeDbName(Option); impl Header for GreptimeDbName { diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 20d6f6dde7f..8ac98bf22f6 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -189,6 +189,7 @@ macro_rules! http_tests { test_pipeline_api, test_pipeline_name_in_header, + test_json_stream_ingestion, test_plain_text_ingestion, test_pipeline_auto_transform, test_pipeline_auto_transform_with_select, @@ -8023,7 +8024,7 @@ transform: assert_eq!( body["error"], json!( - "Invalid request parameter: invalid content type: application/yaml, expected: one of application/json, application/x-ndjson, text/plain" + "Invalid request parameter: invalid content type: application/yaml, expected: one of application/json, application/x-json-stream, application/x-ndjson, text/plain" ) ); @@ -8180,6 +8181,59 @@ transform: guard.remove_all().await; } +pub async fn test_json_stream_ingestion(store_type: StorageType) { + common_telemetry::init_default_ut_logging(); + let (app, mut guard) = + setup_test_http_app_with_frontend(store_type, "test_json_stream_ingestion").await; + + // handshake + let client = TestClient::new(app).await; + + let pipeline_body = r#" +processors: + - date: + field: time + formats: + - "%Y-%m-%d %H:%M:%S%.3f" +transform: + - field: message + type: string + - field: time + type: time + index: timestamp +"#; + + // 1. create pipeline + let res = client + .post("/v1/pipelines/test") + .header("Content-Type", "application/x-yaml") + .body(pipeline_body) + .send() + .await; + assert_eq!(res.status(), StatusCode::OK); + + // 2. write data as concatenated JSON (the `json_stream` format of fluentbit) + let data_body = r#"{"time": "2024-05-25 20:16:37.217", "message": "hello"}{"time": "2024-05-25 20:16:37.218", "message": "hello world"}"#; + let res = client + .post("/v1/ingest?db=public&table=logs1&pipeline_name=test") + .header("Content-Type", "application/x-json-stream") + .body(data_body) + .send() + .await; + assert_eq!(res.status(), StatusCode::OK); + + // 3. check data + validate_data( + "json_stream_ingestion", + &client, + "select message, time from logs1", + r#"[["hello",1716668197217000000],["hello world",1716668197218000000]]"#, + ) + .await; + + guard.remove_all().await; +} + pub async fn test_pipeline_auto_transform(store_type: StorageType) { common_telemetry::init_default_ut_logging(); let (app, mut guard) =