rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-21 19:32:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(servers): accept application/x-json-stream for log ingestion (#9581)

Browse Source 
* 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>
This commit is contained in:
YingNing
2026-10-16 02:36:47 +00:00
committed by GitHub
parent 4859e57e0b
commit ce428bb74d
3 changed files with 119 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/servers/src/http/event.rs
+61 -5
View File
@@ -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<Vec<VrlValue>> {
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#"
src/servers/src/http/header.rs
+3
View File
@@ -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<String>);
impl Header for GreptimeDbName {
tests-integration/tests/http.rs
+55 -1
View File
@@ -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) =