diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs index 3713e99454..e85be71e8a 100644 --- a/src/servers/src/http.rs +++ b/src/servers/src/http.rs @@ -978,6 +978,13 @@ impl HttpServer { "/services/collector/event/1.0", routing::post(splunk::handle_event), ) + // The raw endpoint (plain-text body, one event per line) plus its + // versioned alias. + .route("/services/collector/raw", routing::post(splunk::handle_raw)) + .route( + "/services/collector/raw/1.0", + routing::post(splunk::handle_raw), + ) .layer( ServiceBuilder::new() .layer(RequestDecompressionLayer::new().pass_through_unaccepted(true)), diff --git a/src/servers/src/http/splunk.rs b/src/servers/src/http/splunk.rs index b71b4c6042..2fd16ef1fa 100644 --- a/src/servers/src/http/splunk.rs +++ b/src/servers/src/http/splunk.rs @@ -24,7 +24,7 @@ use std::time::Instant; use api::v1::SemanticType; use axum::Extension; use axum::extract::{Query, State}; -use axum::http::{HeaderMap, StatusCode}; +use axum::http::{HeaderMap, StatusCode, header}; use axum::response::IntoResponse; use bytes::Bytes; use chrono::{DateTime, Utc}; @@ -33,6 +33,7 @@ use common_error::ext::ErrorExt; use common_query::prelude::greptime_timestamp; use common_telemetry::{debug, error}; use operator::insert::SPLUNK_PK_METADATA_ORDER_KEY; +use pipeline::util::to_pipeline_version; use pipeline::{ ContextReq, GREPTIME_INTERNAL_IDENTITY_PIPELINE_NAME, GreptimePipelineParams, PipelineContext, PipelineDefinition, @@ -59,6 +60,89 @@ const DEFAULT_SPLUNK_TABLE: &str = "splunk_logs"; /// `{"text":"HEC is healthy","code":17}`. const HEC_HEALTHY_CODE: u32 = 17; +/// Query parameters for `/services/collector/raw`. +/// `channel` is accepted but ignored until indexer acknowledgment lands. `table`, +/// `pipeline_name`, `version`, and `linebreaker` are Greptime extensions +/// (`linebreaker` opts into event breaking; without it the body is one event). +#[derive(Debug, Default, serde::Deserialize)] +pub struct SplunkRawQueryParams { + pub channel: Option, + pub host: Option, + pub source: Option, + pub sourcetype: Option, + pub index: Option, + pub time: Option, + pub table: Option, + pub pipeline_name: Option, + pub version: Option, + pub linebreaker: Option, +} + +/// Splits a raw body into events. Without `?linebreaker=`, the whole body is ONE +/// event. With `?linebreaker=` (percent-encoded, e.g. `%0A` for `\n`), +/// the body is split on that literal delimiter; whitespace-only segments are +/// dropped, segment content is kept verbatim. +fn split_raw_body<'a>(body: &'a str, linebreaker: Option<&str>) -> Vec<&'a str> { + match linebreaker { + Some(lb) if !lb.is_empty() => body + .split(lb) + .filter(|segment| !segment.trim().is_empty()) + .collect(), + _ => { + if body.trim().is_empty() { + vec![] + } else { + vec![body] + } + } + } +} + +/// Column holding the verbatim raw body on `/raw` (Splunk's `_raw`). Named `message` +/// to avoid clashing with `/event`'s `event` column (whose shape +/// varies by client: string vs identity-flattened object). +const RAW_MESSAGE_COLUMN: &str = "message"; + +/// Collects request-level `/raw` metadata (`host`/`source`/`sourcetype`) present in +/// the query params. The keys double as the tag-column names; values apply to every +/// event in the request (HEC `/raw` metadata is request-level, unlike `/event`). +fn raw_metadata(params: &SplunkRawQueryParams) -> Vec<(&'static str, Bytes)> { + [ + ("host", ¶ms.host), + ("source", ¶ms.source), + ("sourcetype", ¶ms.sourcetype), + ] + .into_iter() + .filter_map(|(key, value)| { + value + .as_deref() + .map(|v| (key, Bytes::copy_from_slice(v.as_bytes()))) + }) + .collect() +} + +/// Maps one raw event to a per-event map: `{ greptime_timestamp: ts, message: , +/// }`. The event text is stored as it is. +fn raw_event_to_map( + event: &str, + ts: DateTime, + metadata: &[(&'static str, Bytes)], +) -> VrlValue { + let mut map: BTreeMap = BTreeMap::new(); + map.insert( + KeyString::from(greptime_timestamp()), + VrlValue::Timestamp(ts), + ); + map.insert( + KeyString::from(RAW_MESSAGE_COLUMN), + VrlValue::Bytes(Bytes::copy_from_slice(event.as_bytes())), + ); + for (key, value) in metadata { + map.insert(KeyString::from(*key), VrlValue::Bytes(value.clone())); + } + VrlValue::Object(map) +} + /// HEC response body `{"text", "code"}`; clients branch on `code`. fn hec_response(status: StatusCode, code: u32, text: &str) -> axum::response::Response { (status, axum::Json(json!({ "text": text, "code": code }))).into_response() @@ -349,14 +433,128 @@ pub async fn handle_event( return hec_response(StatusCode::BAD_REQUEST, 7, &msg); } - // Pipeline: identity by default; override via `pipeline` param or header. - let pipeline_name = params.pipeline_name.clone().unwrap_or_else(|| { + resolve_pipeline_and_ingest( + log_state, + query_ctx, + &headers, + params.pipeline_name.clone(), + params.version.clone(), + requests, + tag_columns, + ) + .await +} + +/// `POST /services/collector/raw` (+ `/raw/1.0` alias). By default, the whole body is +/// raw text stored verbatim as ONE event in the [`RAW_MESSAGE_COLUMN`] — multiline +/// payloads (e.g. stack traces) are preserved intact. Explicit framing is opt-in +/// via `?linebreaker=` (see [`split_raw_body`]). Metadata comes from query params +/// and applies to every event. `channel` (param or `x-splunk-request-channel` header) +/// is accepted but ignored until indexer acknowledgment lands; +#[axum_macros::debug_handler] +pub async fn handle_raw( + State(log_state): State, + Query(params): Query, + Extension(mut query_ctx): Extension, + headers: HeaderMap, + payload: Bytes, +) -> impl IntoResponse { + query_ctx.set_channel(Channel::Splunk); + + // The decompression layer runs strips `Content-Encoding` when it decompresses, + // so a non-identity value means the body is still compressed + if let Some(encoding) = headers.get(header::CONTENT_ENCODING) + && encoding.as_bytes() != b"identity" + { + // HEC code 6 == "invalid data format". + return hec_response(StatusCode::BAD_REQUEST, 6, "invalid data format"); + } + + let Ok(body) = std::str::from_utf8(&payload) else { + debug!("splunk raw body contains invalid UTF-8; rejecting"); + // HEC code 6 == "invalid data format". + return hec_response(StatusCode::BAD_REQUEST, 6, "invalid data format"); + }; + let events = split_raw_body(body, params.linebreaker.as_deref()); + if events.is_empty() { + // HEC code 5 == "No data". + return hec_response(StatusCode::BAD_REQUEST, 5, "No data"); + } + + // Request-level default timestamp: `?time=` (epoch) or ingest time. Splunk would + // additionally extract per-event timestamps from line content; this doesn't support that yet. + let ts = match ¶ms.time { + Some(t) => match parse_hec_time(&VrlValue::Bytes(Bytes::from(t.clone()))) { + Some(ts) => ts, + // HEC code 6 == "invalid data format". + None => return hec_response(StatusCode::BAD_REQUEST, 6, "invalid data format"), + }, + None => Utc::now(), + }; + + // Table routing: `?index=` (sanitized) -> `?table=` -> default. + let table = params + .index + .as_deref() + .and_then(sanitize_index) + .or_else(|| params.table.clone()) + .unwrap_or_else(|| DEFAULT_SPLUNK_TABLE.to_string()); + // Bad table name (e.g. invalid `?table=`) -> HEC code 7. + if !NAME_PATTERN_REG.is_match(&table) { + let msg = format!("Invalid index name: {table}"); + return hec_response(StatusCode::BAD_REQUEST, 7, &msg); + } + + let metadata = raw_metadata(¶ms); + let values = events + .iter() + .map(|event| raw_event_to_map(event, ts, &metadata)) + .collect(); + let tag_columns = HashMap::from([( + table.clone(), + metadata.iter().map(|(key, _)| key.to_string()).collect(), + )]); + let requests = vec![PipelineIngestRequest { table, values }]; + + resolve_pipeline_and_ingest( + log_state, + query_ctx, + &headers, + params.pipeline_name.clone(), + params.version.clone(), + requests, + tag_columns, + ) + .await +} + +/// Shared tail of `/event` and `/raw`: resolves the pipeline (identity default; +/// overridable via param/header, with an optional `?version=` pin), enables tag +/// promotion + metadata-first primary-key ordering for the identity path only, runs +/// the ingest, and maps the outcome to a HEC response. +#[allow(clippy::too_many_arguments)] +async fn resolve_pipeline_and_ingest( + log_state: LogState, + mut query_ctx: QueryContext, + headers: &HeaderMap, + pipeline_name: Option, + version: Option, + requests: Vec, + tag_columns: HashMap>, +) -> axum::response::Response { + // Pipeline: identity by default; override via `pipeline_name` param or header. + let pipeline_name = pipeline_name.unwrap_or_else(|| { headers .get(GREPTIME_PIPELINE_NAME_HEADER_NAME) .and_then(|v| v.to_str().ok()) .unwrap_or(GREPTIME_INTERNAL_IDENTITY_PIPELINE_NAME) .to_string() }); + let version = match to_pipeline_version(version.as_deref()) { + Ok(version) => version, + // HEC code 6 == "invalid data format" (bad `?version=`). + Err(_) => return hec_response(StatusCode::BAD_REQUEST, 6, "invalid pipeline version"), + }; // Only post-process tags for the identity default; respect a user pipeline's schema. let apply_tags = pipeline_name == GREPTIME_INTERNAL_IDENTITY_PIPELINE_NAME; if apply_tags { @@ -366,12 +564,15 @@ pub async fn handle_event( } // custom_time_index so timestamp doesn't get overridden by identity pipeline. let custom_time_index = Some((format!("{};epoch;ns", greptime_timestamp()), false)); - let pipeline = match PipelineDefinition::from_name(&pipeline_name, None, custom_time_index) { + let pipeline = match PipelineDefinition::from_name(&pipeline_name, version, custom_time_index) { Ok(pipeline) => pipeline, - Err(_) => return hec_response(StatusCode::INTERNAL_SERVER_ERROR, 8, "pipeline error"), + Err(e) => { + error!(e; "failed to resolve splunk pipeline definition: {pipeline_name}"); + return hec_response(StatusCode::INTERNAL_SERVER_ERROR, 8, "pipeline error"); + } }; let pipeline_params = - GreptimePipelineParams::from_map(extract_pipeline_params_map_from_headers(&headers)); + GreptimePipelineParams::from_map(extract_pipeline_params_map_from_headers(headers)); match ingest_events( log_state.log_handler, @@ -454,6 +655,116 @@ mod tests { assert!(parse_hec_events(br#"{"event":"a"}{bad}"#).is_err()); } + // ---- split_raw_body ---- + + #[test] + fn splits_raw_body_only_with_explicit_linebreaker() { + // default (no linebreaker): whole body is one event, verbatim. + assert_eq!(split_raw_body("a\nb\r\nc\n", None), vec!["a\nb\r\nc\n"]); + // empty / whitespace-only body -> no events (HEC code 5 upstream). + assert!(split_raw_body("", None).is_empty()); + assert!(split_raw_body(" \n \r\n ", None).is_empty()); + // empty linebreaker behaves like none. + assert_eq!(split_raw_body("a\nb", Some("")), vec!["a\nb"]); + + // explicit "\n": split; whitespace-only segments dropped; + // (a "\r\n"-separated body keeps the "\r" — pass "\r\n" to strip it). + assert_eq!(split_raw_body("a\nb\n", Some("\n")), vec!["a", "b"]); + assert_eq!( + split_raw_body("a\n\n \n\t\nb", Some("\n")), + vec!["a", "b"] + ); + assert_eq!( + split_raw_body("line one\n indented stack frame", Some("\n")), + vec!["line one", " indented stack frame"] + ); + assert_eq!(split_raw_body("a\r\nb", Some("\r\n")), vec!["a", "b"]); + // multi-char literal delimiters work too. + assert_eq!(split_raw_body("a::b::c", Some("::")), vec!["a", "b", "c"]); + // whitespace-only after split -> no events. + assert!(split_raw_body("\n \n", Some("\n")).is_empty()); + } + + // ---- raw_metadata / raw_event_to_map ---- + + #[test] + fn multiline_raw_body_is_one_event() { + // `/raw` must NOT split on newlines unless query parameter is provided. + let stack_trace = "java.lang.NullPointerException: boom\n\ + \tat com.example.Foo.bar(Foo.java:42)\n\ + \tat com.example.Main.main(Main.java:7)"; + let ts = DateTime::from_timestamp(1447828325, 0).unwrap(); + let VrlValue::Object(m) = raw_event_to_map(stack_trace, ts, &[]) else { + panic!("expected object"); + }; + assert_eq!( + m.get(RAW_MESSAGE_COLUMN), + Some(&VrlValue::from(json!(stack_trace))) + ); + } + + #[test] + fn maps_raw_line_with_request_metadata() { + let params = SplunkRawQueryParams { + host: Some("web-01".to_string()), + sourcetype: Some("access_log".to_string()), + ..Default::default() + }; + let meta = raw_metadata(¶ms); + // present params only; keys are the tag-column names. + assert_eq!( + meta, + vec![ + ("host", Bytes::from_static(b"web-01")), + ("sourcetype", Bytes::from_static(b"access_log")) + ] + ); + + let ts = DateTime::from_timestamp(1447828325, 0).unwrap(); + let VrlValue::Object(m) = raw_event_to_map("GET /api 200", ts, &meta) else { + panic!("expected object"); + }; + assert_eq!( + m.get(RAW_MESSAGE_COLUMN), + Some(&VrlValue::from(json!("GET /api 200"))) + ); + assert_eq!(m.get("host"), Some(&VrlValue::from(json!("web-01")))); + assert_eq!( + m.get("sourcetype"), + Some(&VrlValue::from(json!("access_log"))) + ); + // absent metadata (`source`) makes no column. + assert!(!m.contains_key("source")); + assert!(matches!( + m.get(greptime_timestamp()), + Some(VrlValue::Timestamp(dt)) if dt.timestamp() == 1447828325 + )); + + // no query params at all -> just timestamp + message. + let default_params = SplunkRawQueryParams::default(); + let empty = raw_metadata(&default_params); + assert!(empty.is_empty()); + let VrlValue::Object(m) = raw_event_to_map("x", ts, &empty) else { + panic!("expected object"); + }; + assert_eq!(m.len(), 2); + } + + #[test] + fn parses_real_raw_client_payloads() { + // Shapes captured from Vector's `splunk_hec_logs` sink with + // `endpoint_target = "raw"` + let single = "190.79.85.36 - b0rnc0nfused [13/Jul/2026:05:10:25 +0000] \"GET /money HTTP/2.0\" 300 29099"; + let ts = DateTime::from_timestamp(1447828325, 0).unwrap(); + let VrlValue::Object(m) = raw_event_to_map(single, ts, &[]) else { + panic!("expected object"); + }; + assert_eq!( + m.get(RAW_MESSAGE_COLUMN), + Some(&VrlValue::from(json!(single))) + ); + } + // ---- parse_hec_time ---- #[test] diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 5f4e5886d8..f6af868239 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -168,6 +168,7 @@ macro_rules! http_tests { test_splunk_health, test_splunk_health_is_public, test_splunk_logs, + test_splunk_raw, test_log_query, test_jaeger_query_api, test_jaeger_query_api_for_trace_v1, @@ -1715,6 +1716,268 @@ transform: guard.remove_all().await; } +pub async fn test_splunk_raw(store_type: StorageType) { + common_telemetry::init_default_ut_logging(); + + let user_provider = + user_provider_from_option("static_user_provider:cmd:greptime_user=greptime_pwd").unwrap(); + let (app, mut guard) = setup_test_http_app_with_frontend_and_user_provider( + store_type, + "test_splunk_raw", + Some(user_provider), + ) + .await; + let client = TestClient::new(app).await; + + async fn query(client: &TestClient, sql: &str) -> String { + let res = client + .get(format!("/v1/sql?sql={sql}").as_str()) + .header("Authorization", basic_auth("greptime_user", "greptime_pwd")) + .send() + .await; + assert_eq!(res.status(), StatusCode::OK, "query failed: {sql}"); + res.text().await + } + + // HEC `Authorization: Splunk ` + plain-text content type (raw bodies). + let splunk_headers = || { + vec![ + ( + HeaderName::from_static("authorization"), + HeaderValue::from_static("Splunk greptime_user:greptime_pwd"), + ), + ( + HeaderName::from_static("content-type"), + HeaderValue::from_static("text/plain"), + ), + ] + }; + let raw_path = "/v1/splunk/services/collector/raw"; + + // 1. Explicit `?linebreaker=%0A` ("\n") splits the body into one event per + // line, with request-level metadata; `channel` (param AND header) is + // accepted and ignored (ack protocol not implemented); `?time=` sets the + // timestamp for every event; `index` routes the table. Blank lines are + // skipped; indentation inside a line is preserved verbatim. + let mut headers = splunk_headers(); + headers.push(( + HeaderName::from_static("x-splunk-request-channel"), + HeaderValue::from_static("FE0ECFAD-13D5-401B-847D-77833BD77131"), + )); + let body = "line one\nline two\n\n indented line"; + let res = send_req( + &client, + headers, + &format!( + "{raw_path}?channel=FE0ECFAD-13D5-401B-847D-77833BD77131\ + &host=web-01&source=nginx&sourcetype=access&index=raw_main&time=1700000100\ + &linebreaker=%0A" + ), + body.as_bytes().to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + assert!(res.text().await.contains("\"code\":0")); + + // 2. Rows landed: `message` holds each line verbatim, metadata columns carry the + // query-param values, `?time=` (epoch seconds) became the timestamp. + let rows = get_rows_from_output( + &query( + &client, + "select host, source, sourcetype, message, greptime_timestamp from raw_main order by message", + ) + .await, + ); + assert_eq!( + rows, + concat!( + r#"[["web-01","nginx","access"," indented line",1700000100000000000],"#, + r#"["web-01","nginx","access","line one",1700000100000000000],"#, + r#"["web-01","nginx","access","line two",1700000100000000000]]"# + ) + ); + + // 3. host/source/sourcetype are tags (primary key); `message` is not. + let create = query(&client, "show create table raw_main").await; + let pk = create + .split("PRIMARY KEY") + .nth(1) + .expect("raw_main should have a PRIMARY KEY"); + for col in ["host", "source", "sourcetype"] { + assert!( + pk.contains(col), + "expected `{col}` in primary key: {create}" + ); + } + assert!( + !pk.contains("message"), + "`message` must not be a tag: {create}" + ); + + // 4. No query params at all: default table (`splunk_logs`), only timestamp + + // `message` columns, ingest-time timestamp. + let res = send_req( + &client, + splunk_headers(), + raw_path, + b"Hello World".to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + let rows = get_rows_from_output(&query(&client, "select message from splunk_logs").await); + assert_eq!(rows, r#"[["Hello World"]]"#); + + // 4b. Without `?linebreaker=`, a multiline body (e.g. a stack trace) is stored + // as it is. + let stack_trace = "java.lang.NullPointerException: boom\n\tat com.example.Foo.bar(Foo.java:42)\n\tat com.example.Main.main(Main.java:7)"; + let res = send_req( + &client, + splunk_headers(), + &format!("{raw_path}?table=raw_multiline"), + stack_trace.as_bytes().to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + let rows = get_rows_from_output(&query(&client, "select message from raw_multiline").await); + assert_eq!( + rows, + r#"[["java.lang.NullPointerException: boom\n\tat com.example.Foo.bar(Foo.java:42)\n\tat com.example.Main.main(Main.java:7)"]]"# + ); + + // 5. gzip-compressed raw body on the versioned alias (exercises the + // decompression layer and the `/raw/1.0` route). + let res = send_req( + &client, + splunk_headers(), + "/v1/splunk/services/collector/raw/1.0?table=raw_gzip", + b"compressed line".to_vec(), + true, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + let rows = get_rows_from_output(&query(&client, "select message from raw_gzip").await); + assert_eq!(rows, r#"[["compressed line"]]"#); + + // 6. Error paths: empty body -> 5 ("No data"); unparsable `?time=` -> 6; + // invalid `?table=` -> 7 ("incorrect index"). + let res = send_req( + &client, + splunk_headers(), + raw_path, + b" \n ".to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::BAD_REQUEST, res.status()); + assert!(res.text().await.contains("\"code\":5")); + + let res = send_req( + &client, + splunk_headers(), + &format!("{raw_path}?time=not-a-time"), + b"x".to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::BAD_REQUEST, res.status()); + assert!(res.text().await.contains("\"code\":6")); + + let res = send_req( + &client, + splunk_headers(), + &format!("{raw_path}?table=bad%20name"), + b"x".to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::BAD_REQUEST, res.status()); + assert!(res.text().await.contains("\"code\":7")); + + // 7. Replay of a real Vector `splunk_hec_logs` (endpoint_target = "raw") request, + // path and headers verbatim from a wire capture (body trimmed): + let mut headers = splunk_headers(); + headers.push(( + HeaderName::from_static("x-splunk-request-channel"), + HeaderValue::from_static("b408271e-51af-43c1-a99f-9c21f78df0cf"), + )); + let res = send_req( + &client, + headers, + "/v1/splunk/services/collector/raw?source=vector%2Dsrc&sourcetype=vector%5Fdemo&index=vector_raw&host=localhost", + b"245.158.191.1 - AnthraX [13/Jul/2026:05:10:26 +0000] \"GET /wp-admin HTTP/1.0\" 300 17922".to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + let rows = get_rows_from_output( + &query( + &client, + "select host, source, sourcetype, message from vector_raw", + ) + .await, + ); + // the percent-encoded params (`vector%2Dsrc` etc.) decode to the plain values. + assert_eq!( + rows, + r#"[["localhost","vector-src","vector_demo","245.158.191.1 - AnthraX [13/Jul/2026:05:10:26 +0000] \"GET /wp-admin HTTP/1.0\" 300 17922"]]"# + ); + + // 8. An unknown `Content-Encoding` passes through the decompression layer + // still compressed (`pass_through_unaccepted(true)`); the handler must reject + // it (code 6) instead of ingesting compressed bytes. `identity` is fine. + // (Known-but-broken encodings, e.g. `zstd` over gzip bytes, already fail in + // the layer itself.) + // Invalid UTF-8 must also be rejected with code 6, not lossily replaced. + let mut headers = splunk_headers(); + headers.push(( + HeaderName::from_static("content-encoding"), + HeaderValue::from_static("snappy"), + )); + let res = send_req( + &client, + headers, + &format!("{raw_path}?table=raw_encoding"), + compress_vec_with_gzip(b"still compressed".to_vec()), + false, + ) + .await; + assert_eq!(StatusCode::BAD_REQUEST, res.status()); + assert!(res.text().await.contains("\"code\":6")); + + let mut headers = splunk_headers(); + headers.push(( + HeaderName::from_static("content-encoding"), + HeaderValue::from_static("identity"), + )); + let res = send_req( + &client, + headers, + &format!("{raw_path}?table=raw_encoding"), + b"identity line".to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + let rows = get_rows_from_output(&query(&client, "select message from raw_encoding").await); + assert_eq!(rows, r#"[["identity line"]]"#); + + let res = send_req( + &client, + splunk_headers(), + &format!("{raw_path}?table=raw_bad_utf8"), + vec![b'h', b'i', 0xff, 0xfe], + false, + ) + .await; + assert_eq!(StatusCode::BAD_REQUEST, res.status()); + assert!(res.text().await.contains("\"code\":6")); + + guard.remove_all().await; +} + pub async fn test_health_api(store_type: StorageType) { common_telemetry::init_default_ut_logging(); let (app, _guard) = setup_test_http_app_with_frontend(store_type, "health_api").await;