mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-08-18 12:08:22 +00:00
feat: support splunk HEC raw endpoint (#8491)
* feat: add /raw handler + unit tests Signed-off-by: yash <yasha4658@gmail.com> * feat: add integration test for splunk /raw handler Signed-off-by: yash <yasha4658@gmail.com> * fix: added logs and captured payload unit test for splunk /raw endpoint Signed-off-by: yash <yasha4658@gmail.com> * fix: apply cargo fmt and optimize raw_line_to_map Signed-off-by: yash <yasha4658@gmail.com> * fix: preserve entire body as a event with opt-in linebreaker query param Signed-off-by: yash <yasha4658@gmail.com> --------- Signed-off-by: yash <yasha4658@gmail.com>
This commit is contained in:
Yash Agrawal
GitHub
@@ -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)),
|
||||
|
||||
@@ -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<String>,
|
||||
pub host: Option<String>,
|
||||
pub source: Option<String>,
|
||||
pub sourcetype: Option<String>,
|
||||
pub index: Option<String>,
|
||||
pub time: Option<String>,
|
||||
pub table: Option<String>,
|
||||
pub pipeline_name: Option<String>,
|
||||
pub version: Option<String>,
|
||||
pub linebreaker: Option<String>,
|
||||
}
|
||||
|
||||
/// Splits a raw body into events. Without `?linebreaker=`, the whole body is ONE
|
||||
/// event. With `?linebreaker=<literal>` (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: <event>,
|
||||
/// <metadata columns> }`. The event text is stored as it is.
|
||||
fn raw_event_to_map(
|
||||
event: &str,
|
||||
ts: DateTime<Utc>,
|
||||
metadata: &[(&'static str, Bytes)],
|
||||
) -> VrlValue {
|
||||
let mut map: BTreeMap<KeyString, VrlValue> = 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<LogState>,
|
||||
Query(params): Query<SplunkRawQueryParams>,
|
||||
Extension(mut query_ctx): Extension<QueryContext>,
|
||||
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<String>,
|
||||
version: Option<String>,
|
||||
requests: Vec<PipelineIngestRequest>,
|
||||
tag_columns: HashMap<String, HashSet<String>>,
|
||||
) -> 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]
|
||||
|
||||
@@ -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 <user:pass>` + 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;
|
||||
|
||||
Reference in New Issue
Block a user