diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index 0c627af08f..8a038b8013 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -877,6 +877,11 @@ impl Inserter { let mut create_table_expr = build_create_table_expr(&table_ref, request_schema, engine_name)?; + // extension set by the Splunk HEC handler for identity path + if ctx.extension(SPLUNK_PK_METADATA_ORDER_KEY).is_some() { + reorder_splunk_primary_keys(&mut create_table_expr.primary_keys); + } + info!("Table `{table_ref}` does not exist, try creating table"); create_table_expr.table_options.extend(table_options); Ok(create_table_expr) @@ -1188,6 +1193,24 @@ pub fn build_create_table_expr( expr_helper::create_table_expr_by_column_schemas(table, request_schema, engine, None) } +/// `QueryContext` extension key the Splunk HEC handler sets (to `"true"`) on its identity +/// path to request metadata-first primary-key ordering at table creation. It is absent for +/// user-supplied pipelines, so their primary-key order is left untouched. +pub const SPLUNK_PK_METADATA_ORDER_KEY: &str = "splunk_pk_metadata_order"; + +/// Moves Splunk's metadata tags (`host`, `source`, `sourcetype`) to the front of the +/// primary key, keeping the relative order of the remaining tags. +fn reorder_splunk_primary_keys(primary_keys: &mut [String]) { + const LEAD: [&str; 3] = ["host", "source", "sourcetype"]; + // Stable sort: `LEAD` columns move to the front in `host`/`source`/`sourcetype` order; + // every other column keeps its existing relative position. + primary_keys.sort_by_key(|name| { + LEAD.iter() + .position(|&lead| lead == name.as_str()) + .unwrap_or(LEAD.len()) + }); +} + /// Result of `create_or_alter_tables_on_demand`. struct CreateAlterTableResult { /// table ids of ttl=instant tables. diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs index 900ce067cb..5f5462f264 100644 --- a/src/servers/src/http.rs +++ b/src/servers/src/http.rs @@ -104,6 +104,7 @@ pub mod pprof; pub mod prom_store; pub mod prometheus; pub mod result; +pub mod splunk; mod timeout; pub mod utils; @@ -127,8 +128,12 @@ const DEFAULT_BODY_LIMIT: ReadableSize = ReadableSize::mb(64); pub const AUTHORIZATION_HEADER: &str = "x-greptime-auth"; // TODO(fys): This is a temporary workaround, it will be improved later -pub static PUBLIC_API_PREFIX: [&str; 3] = - ["/v1/influxdb/ping", "/v1/influxdb/health", "/v1/health"]; +pub static PUBLIC_API_PREFIX: [&str; 4] = [ + "/v1/influxdb/ping", + "/v1/influxdb/health", + "/v1/health", + "/v1/splunk/services/collector/health", +]; #[derive(Default)] pub struct HttpServer { @@ -674,7 +679,12 @@ impl HttpServerBuilder { &format!("/{HTTP_API_VERSION}/elasticsearch/"), Router::new() .route("/", routing::get(elasticsearch::handle_get_version)) - .with_state(log_state), + .with_state(log_state.clone()), + ); + + let router = router.nest( + &format!("/{HTTP_API_VERSION}/splunk"), + HttpServer::route_splunk(log_state), ); Self { router, ..self } @@ -931,6 +941,34 @@ impl HttpServer { .with_state(log_state) } + fn route_splunk(log_state: LogState) -> Router { + Router::new() + .route( + "/services/collector/health", + routing::get(splunk::handle_health), + ) + .route( + "/services/collector/health/1.0", + routing::get(splunk::handle_health), + ) + // The event endpoint plus its base and versioned aliases all serve + // the same handler (Splunk JSON event protocol). + .route( + "/services/collector/event", + routing::post(splunk::handle_event), + ) + .route("/services/collector", routing::post(splunk::handle_event)) + .route( + "/services/collector/event/1.0", + routing::post(splunk::handle_event), + ) + .layer( + ServiceBuilder::new() + .layer(RequestDecompressionLayer::new().pass_through_unaccepted(true)), + ) + .with_state(log_state) + } + fn route_elasticsearch(log_state: LogState) -> Router { Router::new() // Return fake responsefor HEAD '/' request. diff --git a/src/servers/src/http/authorize.rs b/src/servers/src/http/authorize.rs index 2559ba11a7..a6d8ce2438 100644 --- a/src/servers/src/http/authorize.rs +++ b/src/servers/src/http/authorize.rs @@ -33,10 +33,12 @@ use snafu::{OptionExt, ResultExt, ensure}; use crate::error::{ self, InvalidAuthHeaderInvisibleASCIISnafu, InvalidAuthHeaderSnafu, InvalidParameterSnafu, - NotFoundInfluxAuthSnafu, Result, UnsupportedAuthSchemeSnafu, UrlDecodeSnafu, + NotFoundAuthHeaderSnafu, NotFoundInfluxAuthSnafu, Result, UnsupportedAuthSchemeSnafu, + UrlDecodeSnafu, }; use crate::http::header::{GREPTIME_TIMEZONE_HEADER_NAME, GreptimeDbName}; use crate::http::result::error_result::ErrorResponse; +use crate::http::splunk::is_splunk_request; use crate::http::{AUTHORIZATION_HEADER, HTTP_API_PREFIX, PUBLIC_API_PREFIX}; use crate::influxdb::{is_influxdb_request, is_influxdb_v2_request}; @@ -86,6 +88,14 @@ pub async fn inner_auth( crate::metrics::METRIC_AUTH_FAILURE .with_label_values(&[e.status_code().as_ref()]) .inc(); + if is_splunk_request(&req) { + // HEC: missing header -> 2 ("token is required"), else 4 ("invalid token"). + let (status, code) = match &e { + error::Error::NotFoundAuthHeader { .. } => (StatusCode::UNAUTHORIZED, 2), + _ => (StatusCode::FORBIDDEN, 4), + }; + return Err(splunk_hec_err(status, code)); + } return Err(err_response(e)); } }; @@ -110,6 +120,10 @@ pub async fn inner_auth( crate::metrics::METRIC_AUTH_FAILURE .with_label_values(&[e.status_code().as_ref()]) .inc(); + // HEC: bad credentials -> 4 ("invalid token", 403). + if is_splunk_request(&req) { + return Err(splunk_hec_err(StatusCode::FORBIDDEN, 4)); + } Err(err_response(e)) } } @@ -126,6 +140,20 @@ pub async fn check_http_auth( } } +/// HEC-shaped auth error (`{"text","code"}`) so Splunk clients can branch on `code`. +fn splunk_hec_err(status: StatusCode, code: u32) -> Response { + let text = match code { + 2 => "Token is required", + 4 => "Invalid token", + _ => "Unauthorized", + }; + ( + status, + axum::Json(serde_json::json!({ "text": text, "code": code })), + ) + .into_response() +} + fn err_response(err: impl ErrorExt) -> Response { (StatusCode::UNAUTHORIZED, ErrorResponse::from_error(err)).into_response() } @@ -203,10 +231,33 @@ fn get_influxdb_credentials(request: &Request) -> Result(request: &Request) -> Result> { + let Some(header) = request.headers().get(http::header::AUTHORIZATION) else { + return Ok(None); + }; + let (auth_scheme, credential) = header + .to_str() + .context(InvalidAuthHeaderInvisibleASCIISnafu)? + .split_once(' ') + .context(InvalidAuthHeaderSnafu)?; + + let (username, password) = match auth_scheme.to_lowercase().as_str() { + "splunk" => { + let (u, p) = credential.split_once(':').context(InvalidAuthHeaderSnafu)?; + (u.to_string(), p.to_string().into()) + } + "basic" => decode_basic(credential)?, + _ => UnsupportedAuthSchemeSnafu { name: auth_scheme }.fail()?, + }; + Ok(Some((username, password))) +} + pub fn extract_username_and_password(request: &Request) -> Result<(Username, Password)> { Ok(if is_influxdb_request(request) { // compatible with influxdb auth get_influxdb_credentials(request)?.context(NotFoundInfluxAuthSnafu)? + } else if is_splunk_request(request) { + get_splunk_credentials(request)?.context(NotFoundAuthHeaderSnafu)? } else { // normal http auth let scheme = auth_header(request)?; @@ -360,6 +411,59 @@ mod tests { assert!(need_auth(&req)); } + #[test] + fn test_splunk_auth() { + let splunk_uri = "http://127.0.0.1/v1/splunk/services/collector/event"; + let splunk_req = |auth: Option<&str>| { + let mut req = Request::builder().uri(splunk_uri); + if let Some(auth) = auth { + req = req.header(http::header::AUTHORIZATION, auth); + } + req.body(()).unwrap() + }; + + // is_splunk_request matches our mount, not other endpoints. + assert!(is_splunk_request(&splunk_req(None))); + assert!(!is_splunk_request( + &Request::builder() + .uri("http://127.0.0.1/v1/influxdb/write") + .body(()) + .unwrap() + )); + assert!(!is_splunk_request( + &Request::builder() + .uri("http://127.0.0.1/v1/sql") + .body(()) + .unwrap() + )); + + // `Splunk ` -> (user, pass). + let (username, password) = + get_splunk_credentials(&splunk_req(Some("Splunk teamA:secretA"))) + .unwrap() + .unwrap(); + assert_eq!(username, "teamA"); + assert_eq!(password.expose_secret(), "secretA"); + + // standard Basic is also accepted (parity with influxdb). + let basic = basic_auth("u", "p"); + let (username, password) = get_splunk_credentials(&splunk_req(Some(&basic))) + .unwrap() + .unwrap(); + assert_eq!(username, "u"); + assert_eq!(password.expose_secret(), "p"); + + // missing header -> None; token without ':' -> error. + assert!(get_splunk_credentials(&splunk_req(None)).unwrap().is_none()); + assert!(get_splunk_credentials(&splunk_req(Some("Splunk no_colon_token"))).is_err()); + + // full dispatch routes a splunk request through the splunk scheme. + let (username, password) = + extract_username_and_password(&splunk_req(Some("Splunk teamA:secretA"))).unwrap(); + assert_eq!(username, "teamA"); + assert_eq!(password.expose_secret(), "secretA"); + } + #[test] fn test_decode_basic() { let credential = basic_auth_credentials("username", "password"); diff --git a/src/servers/src/http/event.rs b/src/servers/src/http/event.rs index f08c76bcdb..7a5f40840b 100644 --- a/src/servers/src/http/event.rs +++ b/src/servers/src/http/event.rs @@ -384,7 +384,7 @@ pub async fn delete_pipeline( /// Transform NDJSON array into a single array /// always return an array -fn transform_ndjson_array_factory( +pub(crate) fn transform_ndjson_array_factory( values: impl IntoIterator>, ignore_error: bool, ) -> Result> { diff --git a/src/servers/src/http/splunk.rs b/src/servers/src/http/splunk.rs new file mode 100644 index 0000000000..b71b4c6042 --- /dev/null +++ b/src/servers/src/http/splunk.rs @@ -0,0 +1,848 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Splunk HTTP Event Collector (HEC) compatible ingestion endpoint. +//! +//! Clients point their base endpoint at `/v1/splunk`, so the full paths are e.g. +//! `/v1/splunk/services/collector/event` and `/v1/splunk/services/collector/health`. + +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::sync::Arc; +use std::time::Instant; + +use api::v1::SemanticType; +use axum::Extension; +use axum::extract::{Query, State}; +use axum::http::{HeaderMap, StatusCode}; +use axum::response::IntoResponse; +use bytes::Bytes; +use chrono::{DateTime, Utc}; +use common_base::regex_pattern::NAME_PATTERN_REG; +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::{ + ContextReq, GREPTIME_INTERNAL_IDENTITY_PIPELINE_NAME, GreptimePipelineParams, PipelineContext, + PipelineDefinition, +}; +use serde_json::{Deserializer, json}; +use session::context::{Channel, QueryContext, QueryContextRef}; +use vrl::value::{KeyString, Value as VrlValue}; + +use crate::error::{Result, status_code_to_http_status}; +use crate::http::HttpResponse; +use crate::http::event::{ + LogIngesterQueryParams, LogState, PipelineIngestRequest, execute_log_context_req, + extract_pipeline_params_map_from_headers, transform_ndjson_array_factory, +}; +use crate::http::header::constants::GREPTIME_PIPELINE_NAME_HEADER_NAME; +use crate::metrics::{METRIC_HTTP_LOGS_INGESTION_COUNTER, METRIC_HTTP_LOGS_INGESTION_ELAPSED}; +use crate::pipeline::run_pipeline; +use crate::query_handler::PipelineHandlerRef; + +/// Default table used when neither the event's `index` nor a `?table=` query +/// param is provided. +const DEFAULT_SPLUNK_TABLE: &str = "splunk_logs"; +/// HEC response code for a healthy collector. Splunk returns +/// `{"text":"HEC is healthy","code":17}`. +const HEC_HEALTHY_CODE: u32 = 17; + +/// 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() +} + +/// Parses a HEC body into a flat list of events. Handles both batch forms: objects +/// concatenated with any/no separator, and a top-level array (flattened). +fn parse_hec_events(body: &[u8]) -> Result> { + let values = Deserializer::from_slice(body).into_iter::(); + // ignore_error = false: reject the whole batch on a malformed value. + transform_ndjson_array_factory(values, false) +} + +/// HEC `time`: epoch seconds (optionally fractional); values past ~1e12 are read as +/// milliseconds. `None` if absent/unparseable (caller falls back to ingest time). +fn parse_hec_time(value: &VrlValue) -> Option> { + let n: f64 = match value { + VrlValue::Integer(i) => *i as f64, + VrlValue::Float(f) => f.into_inner(), + VrlValue::Bytes(b) => std::str::from_utf8(b).ok()?.trim().parse().ok()?, + VrlValue::Timestamp(dt) => return Some(*dt), + _ => return None, + }; + if !n.is_finite() { + return None; + } + const MILLIS_THRESHOLD: f64 = 1e12; + // Safe (`Option`-returning) constructors: out-of-range input yields `None`, not a panic. + if n >= MILLIS_THRESHOLD { + DateTime::from_timestamp_millis(n as i64) + } else { + let secs = n.floor() as i64; + let nsecs = ((n - n.floor()) * 1e9) as u32; + DateTime::from_timestamp(secs, nsecs) + } +} + +/// `event` missing -> 12, `event` blank -> 13. +/// present, non-null but unparsable `time` -> 6. +fn validate_event(event: &VrlValue) -> Option<(u32, &'static str)> { + let VrlValue::Object(obj) = event else { + return None; + }; + match obj.get("event") { + None => return Some((12, "Event field is required")), + Some(value) if is_blank_event(value) => return Some((13, "Event field cannot be blank")), + _ => {} + } + if let Some(time) = obj.get("time") + && !matches!(time, VrlValue::Null) + && parse_hec_time(time).is_none() + { + return Some((6, "invalid data format")); + } + None +} + +/// A HEC `event` value is blank if it's `null` or an empty/whitespace-only string. +fn is_blank_event(value: &VrlValue) -> bool { + match value { + VrlValue::Null => true, + VrlValue::Bytes(b) => std::str::from_utf8(b).is_ok_and(|s| s.trim().is_empty()), + _ => false, + } +} + +/// Maps one HEC event to `(table, per-event map, tag names)`: `time`->timestamp, +/// `index`->table, host/source/sourcetype/`fields`->tags, `event`+rest->data. +/// `None` if the event isn't a JSON object. +fn hec_event_to_map( + event: VrlValue, + query_table: Option<&str>, +) -> Option<(String, VrlValue, Vec)> { + let mut obj = match event { + VrlValue::Object(obj) => obj, + other => { + debug!("skipping non-object splunk HEC event: {other:?}"); + return None; + } + }; + + // Timestamp: HEC `time` is honored first, else ingest time. + let ts = obj + .remove("time") + .as_ref() + .and_then(parse_hec_time) + .unwrap_or_else(Utc::now); + + // Table routing: `index` (consumed) -> `?table=` -> default. + let index = match obj.remove("index") { + Some(VrlValue::Bytes(b)) => Some(String::from_utf8_lossy(&b).into_owned()), + _ => None, + }; + let table = index + .as_deref() + .and_then(sanitize_index) + .or_else(|| query_table.map(str::to_string)) + .unwrap_or_else(|| DEFAULT_SPLUNK_TABLE.to_string()); + + let mut map: BTreeMap = BTreeMap::new(); + map.insert( + KeyString::from(greptime_timestamp()), + VrlValue::Timestamp(ts), + ); + + let mut tag_names: Vec = Vec::new(); + + // `fields` is flat: spread its keys to top-level columns, all tags. + if let Some(VrlValue::Object(fields)) = obj.remove("fields") { + for (k, v) in fields { + tag_names.push(k.as_str().to_string()); + map.insert(k, v); + } + } + + // host / source / sourcetype are tags. + for key in ["host", "source", "sourcetype"] { + if let Some(v) = obj.remove(key) { + tag_names.push(key.to_string()); + map.insert(KeyString::from(key), v); + } + } + + // `event` and any remaining keys are data columns. + for (k, v) in obj { + map.insert(k, v); + } + + Some((table, VrlValue::Object(map), tag_names)) +} + +/// Retags `Field` columns to `Tag` per table (identity makes everything a Field) so the +/// insert path adds them to the primary key. Tags are scoped by table name so a batch +/// targeting multiple tables can't cross-promote a same-named field. Identity-only: +/// rebuilds under the default opt. +fn apply_tag_columns( + ctx_req: ContextReq, + tag_columns: &HashMap>, +) -> ContextReq { + let mut reqs = ctx_req.all_req().collect::>(); + for req in &mut reqs { + let Some(rows) = req.rows.as_mut() else { + continue; + }; + let Some(tags) = tag_columns.get(&req.table_name) else { + continue; + }; + for col in &mut rows.schema { + if tags.contains(&col.column_name) { + col.semantic_type = SemanticType::Tag as i32; + } + } + } + ContextReq::default_opt_with_reqs(reqs) +} + +/// Coerces a Splunk `index` into a valid table name (`NAME_PATTERN`); `None` if empty. +fn sanitize_index(raw: &str) -> Option { + let trimmed = raw.trim(); + if trimmed.is_empty() { + return None; + } + if NAME_PATTERN_REG.is_match(trimmed) { + return Some(trimmed.to_string()); + } + let mut out = String::with_capacity(trimmed.len()); + for c in trimmed.chars() { + // body-allowed set + if c.is_ascii_alphanumeric() || matches!(c, '_' | ':' | '-' | '.' | '@' | '#') { + out.push(c); + } else { + out.push('_'); // spaces, slashes, unicode, etc. → '_' + } + } + + let first_ok = out + .chars() + .next() + .map(|c| c.is_ascii_alphabetic() || matches!(c, '_' | ':' | '-')) + .unwrap_or(false); + + if !first_ok { + out.insert(0, '_'); + } + + NAME_PATTERN_REG.is_match(&out).then_some(out) +} + +pub(crate) fn is_splunk_request(req: &axum::extract::Request) -> bool { + // Match only `/v1/splunk/` + req.uri().path().starts_with("/v1/splunk/") +} +/// Like `ingest_logs_inner`, but retags metadata columns (identity default) before insert. +async fn ingest_events( + handler: PipelineHandlerRef, + pipeline: PipelineDefinition, + requests: Vec, + query_ctx: QueryContextRef, + pipeline_params: GreptimePipelineParams, + tag_columns: HashMap>, + apply_tags: bool, +) -> Result { + let exec_timer = Instant::now(); + let pipeline_ctx = PipelineContext::new(&pipeline, &pipeline_params, query_ctx.channel()); + + let mut ctx_req = ContextReq::default(); + for req in requests { + ctx_req.merge(run_pipeline(&handler, &pipeline_ctx, req, &query_ctx, true).await?); + } + + let ctx_req = if apply_tags { + apply_tag_columns(ctx_req, &tag_columns) + } else { + ctx_req + }; + + execute_log_context_req( + handler, + ctx_req, + query_ctx, + exec_timer, + &METRIC_HTTP_LOGS_INGESTION_COUNTER, + &METRIC_HTTP_LOGS_INGESTION_ELAPSED, + ) + .await +} + +/// `GET /services/collector/health` (+ `/1.0`). Public (see `PUBLIC_API_PREFIX`), +/// since clients probe it before sending. `ack`/`token` query params are ignored. +#[axum_macros::debug_handler] +pub async fn handle_health() -> impl IntoResponse { + hec_response(StatusCode::OK, HEC_HEALTHY_CODE, "HEC is healthy") +} + +/// `POST /services/collector/event` (+ `/services/collector`, `/event/1.0` aliases). +/// Parses HEC events, runs them through the pipeline (identity default, overridable), +/// and inserts with metadata columns as tags. +#[axum_macros::debug_handler] +pub async fn handle_event( + State(log_state): State, + Query(params): Query, + Extension(mut query_ctx): Extension, + headers: HeaderMap, + payload: Bytes, +) -> impl IntoResponse { + query_ctx.set_channel(Channel::Splunk); + let events = match parse_hec_events(&payload) { + Ok(events) => events, + // HEC code 6 == "invalid data format". + Err(_) => return hec_response(StatusCode::BAD_REQUEST, 6, "invalid data format"), + }; + if events.is_empty() { + // HEC code 5 == "No data". + return hec_response(StatusCode::BAD_REQUEST, 5, "No data"); + } + + // Map each event -> (table, per-event map, tag names); group by table. + let query_table = params.table.as_deref(); + let mut by_table: HashMap> = HashMap::new(); + let mut tag_columns: HashMap> = HashMap::new(); + for event in events { + // Reject the batch on an invalid event: missing/blank `event` (12/13) or an + // unparsable `time` (6). + if let Some((code, text)) = validate_event(&event) { + return hec_response(StatusCode::BAD_REQUEST, code, text); + } + if let Some((table, map, tags)) = hec_event_to_map(event, query_table) { + tag_columns.entry(table.clone()).or_default().extend(tags); + by_table.entry(table).or_default().push(map); + } + } + let requests: Vec = by_table + .into_iter() + .map(|(table, values)| PipelineIngestRequest { table, values }) + .collect(); + + // Events parsed but none were JSON objects, so nothing is ingestable. HEC code 6 == "invalid data format". + if requests.is_empty() { + return hec_response(StatusCode::BAD_REQUEST, 6, "invalid data format"); + } + + // Bad table name (e.g. invalid `?table=`) -> HEC code 7 ("incorrect index"). + if let Some(bad) = requests + .iter() + .find(|r| !NAME_PATTERN_REG.is_match(&r.table)) + { + let msg = format!("incorrect index: {}", bad.table); + 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(|| { + headers + .get(GREPTIME_PIPELINE_NAME_HEADER_NAME) + .and_then(|v| v.to_str().ok()) + .unwrap_or(GREPTIME_INTERNAL_IDENTITY_PIPELINE_NAME) + .to_string() + }); + // 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 { + // Ask the insert path to lead the primary key with our metadata tags. Scoped to the + // identity path so a user pipeline's own key order is left untouched. + query_ctx.set_extension(SPLUNK_PK_METADATA_ORDER_KEY, "true"); + } + // 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) { + Ok(pipeline) => pipeline, + Err(_) => return hec_response(StatusCode::INTERNAL_SERVER_ERROR, 8, "pipeline error"), + }; + let pipeline_params = + GreptimePipelineParams::from_map(extract_pipeline_params_map_from_headers(&headers)); + + match ingest_events( + log_state.log_handler, + pipeline, + requests, + Arc::new(query_ctx), + pipeline_params, + tag_columns, + apply_tags, + ) + .await + { + // HEC code 0 == "Success". + Ok(_) => hec_response(StatusCode::OK, 0, "Success"), + Err(e) => { + error!(e; "failed to ingest splunk hec events"); + // client errors -> HEC code 6, else 8. + let status = status_code_to_http_status(&e.status_code()); + let code = if status.is_client_error() { 6 } else { 8 }; + let msg = e.to_string(); + hec_response(status, code, &msg) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn events_for(body: &[u8]) -> Vec { + parse_hec_events(body).unwrap() + } + + #[test] + fn parses_single_object() { + let events = events_for(br#"{"event":"hello","time":1}"#); + assert_eq!(events, vec![json!({"event":"hello","time":1}).into()]); + } + + #[test] + fn parses_concatenated_objects_without_separator() { + let events = events_for(br#"{"event":"a"}{"event":"b"}"#); + assert_eq!( + events, + vec![json!({"event":"a"}).into(), json!({"event":"b"}).into()] + ); + } + + #[test] + fn parses_newline_and_whitespace_separated_objects() { + // newline, leading spaces, and a tab between objects — none are required. + let events = events_for(b"{\"event\":\"a\"}\n {\"event\":\"b\"}\t{\"event\":\"c\"}"); + assert_eq!(events.len(), 3); + } + + #[test] + fn parses_top_level_array_into_flat_events() { + let events = events_for(br#"[{"event":"a"},{"event":"b"}]"#); + assert_eq!( + events, + vec![json!({"event":"a"}).into(), json!({"event":"b"}).into()] + ); + } + + #[test] + fn flattens_mixed_array_and_trailing_object() { + // a top-level array immediately followed by a bare object. + let events = events_for(br#"[{"event":"a"},{"event":"b"}]{"event":"c"}"#); + assert_eq!(events.len(), 3); + } + + #[test] + fn empty_or_whitespace_body_yields_no_events() { + assert!(events_for(b"").is_empty()); + assert!(events_for(b" \n ").is_empty()); + } + + #[test] + fn malformed_json_is_rejected() { + assert!(parse_hec_events(br#"{"event":"a"}{bad}"#).is_err()); + } + + // ---- parse_hec_time ---- + + #[test] + fn parse_time_integer_seconds() { + let v: VrlValue = json!(1426279439).into(); + assert_eq!(parse_hec_time(&v).unwrap().timestamp(), 1426279439); + } + + #[test] + fn parse_time_fractional_seconds_keeps_millis() { + let v: VrlValue = json!(1426279439.5).into(); + // the .5s must survive (not be truncated like EpochProcessor would). + assert_eq!( + parse_hec_time(&v).unwrap().timestamp_millis(), + 1426279439500 + ); + } + + #[test] + fn parse_time_integer_millis() { + let v: VrlValue = json!(1447828325000_i64).into(); + // past the millis threshold -> read as ms, same instant as 1447828325s. + assert_eq!(parse_hec_time(&v).unwrap().timestamp(), 1447828325); + } + + #[test] + fn parse_time_string_number() { + let v: VrlValue = json!("1426279439").into(); + assert_eq!(parse_hec_time(&v).unwrap().timestamp(), 1426279439); + } + + #[test] + fn parse_time_passthrough_timestamp() { + let dt = DateTime::from_timestamp_nanos(123_456_789); + assert_eq!(parse_hec_time(&VrlValue::Timestamp(dt)), Some(dt)); + } + + #[test] + fn parse_time_missing_or_invalid_is_none() { + assert!(parse_hec_time(&VrlValue::Null).is_none()); + let not_num: VrlValue = json!("not a number").into(); + assert!(parse_hec_time(¬_num).is_none()); + let obj: VrlValue = json!({ "x": 1 }).into(); + assert!(parse_hec_time(&obj).is_none()); + } + + // ---- sanitize_index ---- + + #[test] + fn sanitize_keeps_valid_names() { + assert_eq!(sanitize_index("main").as_deref(), Some("main")); + assert_eq!( + sanitize_index("web-prod.2024").as_deref(), + Some("web-prod.2024") + ); + assert_eq!( + sanitize_index("cpu:metrics").as_deref(), + Some("cpu:metrics") + ); + } + + #[test] + fn sanitize_replaces_invalid_chars() { + assert_eq!( + sanitize_index("my index/v2").as_deref(), + Some("my_index_v2") + ); + } + + #[test] + fn sanitize_fixes_leading_digit() { + assert_eq!(sanitize_index("123logs").as_deref(), Some("_123logs")); + } + + #[test] + fn sanitize_empty_is_none() { + assert!(sanitize_index("").is_none()); + assert!(sanitize_index(" ").is_none()); + } + + #[test] + fn sanitize_output_is_always_a_valid_table_name() { + // Invariant: a non-empty input never yields a name the create path would reject. + for raw in [ + "main", + "web-prod.2024", + "my index/v2", + "123", + "@#@#", + "...", + "日本語 logs", + "a/b\\c", + ] { + if let Some(name) = sanitize_index(raw) { + assert!( + NAME_PATTERN_REG.is_match(&name), + "sanitized {raw:?} -> {name:?} is not a valid table name" + ); + } + } + } + + // ---- hec_event_to_map ---- + + #[test] + fn map_extracts_metadata_and_routes_by_index() { + let event: VrlValue = json!({ + "time": 1426279439, + "host": "web-01", + "source": "nginx", + "sourcetype": "access", + "index": "web_logs", + "event": "GET /api 200", + "fields": { "region": "us-east" } + }) + .into(); + + let (table, map, tags) = hec_event_to_map(event, None).unwrap(); + + // `index` -> table name. + assert_eq!(table, "web_logs"); + + // tags = host/source/sourcetype + each `fields` key. + let tagset: HashSet<&str> = tags.iter().map(String::as_str).collect(); + assert_eq!( + tagset, + HashSet::from(["host", "source", "sourcetype", "region"]) + ); + + let VrlValue::Object(m) = map else { + panic!("expected object"); + }; + // metadata + fields became columns with their values. + assert_eq!(m.get("host"), Some(&VrlValue::from(json!("web-01")))); + assert_eq!(m.get("region"), Some(&VrlValue::from(json!("us-east")))); + assert_eq!(m.get("event"), Some(&VrlValue::from(json!("GET /api 200")))); + // `time` became the timestamp column, not a `time` column. + assert!(!m.contains_key("time")); + assert!(matches!( + m.get(greptime_timestamp()), + Some(VrlValue::Timestamp(dt)) if dt.timestamp() == 1426279439 + )); + // `index` and `fields` are consumed, not columns. + assert!(!m.contains_key("index")); + assert!(!m.contains_key("fields")); + } + + #[test] + fn map_falls_back_to_query_table_then_default() { + let ev1: VrlValue = json!({ "event": "x" }).into(); + let (t1, _, _) = hec_event_to_map(ev1, Some("from_query")).unwrap(); + assert_eq!(t1, "from_query"); + + let ev2: VrlValue = json!({ "event": "x" }).into(); + let (t2, _, _) = hec_event_to_map(ev2, None).unwrap(); + assert_eq!(t2, "splunk_logs"); + } + + #[test] + fn map_sanitizes_index_for_table() { + let ev: VrlValue = json!({ "index": "web/prod", "event": "x" }).into(); + let (table, _, _) = hec_event_to_map(ev, None).unwrap(); + assert_eq!(table, "web_prod"); + } + + #[test] + fn map_rejects_non_object_event() { + let ev: VrlValue = json!("just a string").into(); + assert!(hec_event_to_map(ev, None).is_none()); + } + + // ---- validate_event ---- + + #[test] + fn validates_event() { + let check = |v: serde_json::Value| validate_event(&v.into()); + + // missing `event` -> code 12. + assert_eq!( + check(json!({ "host": "h" })), + Some((12, "Event field is required")) + ); + // present but blank (empty / whitespace) or null -> code 13. + assert_eq!( + check(json!({ "event": "" })), + Some((13, "Event field cannot be blank")) + ); + assert_eq!( + check(json!({ "event": " " })), + Some((13, "Event field cannot be blank")) + ); + assert_eq!( + check(json!({ "event": null })), + Some((13, "Event field cannot be blank")) + ); + // valid: non-empty string, object, or other non-blank value. + assert_eq!(check(json!({ "event": "hello" })), None); + assert_eq!(check(json!({ "event": { "a": 1 } })), None); + assert_eq!(check(json!({ "event": 0 })), None); + // non-object events aren't validated here (handled by `hec_event_to_map`). + assert_eq!(check(json!("just a string")), None); + + // present but unparsable `time` -> code 6 (number string / numeric are fine). + let bad_time = Some((6, "invalid data format")); + assert_eq!( + check(json!({ "event": "x", "time": "not-a-time" })), + bad_time + ); + assert_eq!(check(json!({ "event": "x", "time": { "a": 1 } })), bad_time); + assert_eq!(check(json!({ "event": "x", "time": 1700000000 })), None); + assert_eq!(check(json!({ "event": "x", "time": "1700000000" })), None); + // absent or null `time` falls back to ingest time, so it's allowed. + assert_eq!(check(json!({ "event": "x" })), None); + assert_eq!(check(json!({ "event": "x", "time": null })), None); + } + + #[test] + fn parses_minimal_events_in_both_batch_forms() { + // Splunk docs "Example 3": minimal events (`event` + `time` only), sent both as + // concatenated objects (whitespace-separated) and as a JSON array. + let concatenated = r#"{ + "event": "event 1", + "time": 1447828325 +} + +{ + "event": "event 2", + "time": 1447828326 +}"#; + let array = r#"[ + { "event": "event 1", "time": 1447828325 }, + { "event": "event 2", "time": 1447828326 } +]"#; + + for body in [concatenated, array] { + let events = parse_hec_events(body.as_bytes()).unwrap(); + assert_eq!(events.len(), 2); + + let (table, map, tags) = + hec_event_to_map(events.into_iter().next().unwrap(), None).unwrap(); + assert_eq!(table, "splunk_logs"); // no `index` -> default table + assert!(tags.is_empty()); // no host/source/sourcetype/fields + let VrlValue::Object(m) = map else { + panic!("expected object"); + }; + assert_eq!(m.get("event"), Some(&VrlValue::from(json!("event 1")))); + assert!(matches!( + m.get(greptime_timestamp()), + Some(VrlValue::Timestamp(dt)) if dt.timestamp() == 1447828325 + )); + assert!(!m.contains_key("host")); + } + } + + #[test] + fn map_keeps_event_object_for_pipeline_flattening() { + let ev: VrlValue = json!({ "event": { "a": 1 } }).into(); + let (_, map, _) = hec_event_to_map(ev, None).unwrap(); + let VrlValue::Object(m) = map else { + panic!("expected object"); + }; + assert!(matches!(m.get("event"), Some(VrlValue::Object(_)))); + } + + #[test] + fn map_uses_ingest_time_when_time_absent() { + let ev: VrlValue = json!({ "event": "x" }).into(); + let (_, map, _) = hec_event_to_map(ev, None).unwrap(); + let VrlValue::Object(m) = map else { + panic!("expected object"); + }; + assert!(matches!( + m.get(greptime_timestamp()), + Some(VrlValue::Timestamp(_)) + )); + } + + // ---- real client payload ---- + + #[test] + fn parses_real_client_payloads() { + // Shapes captured from real `splunk_hec` clients (values trimmed, structure + // verbatim). The two clients deliberately disagree on batch separator, + // `event` type, and `fields` keys — the parser must handle all of it. + + // --- Vector splunk_hec sink: NO separator; `event` is an object. --- + let vector = concat!( + r#"{"event":{"message":"GET /api 200","status":"200"},"fields":{"region":"us-east"},"#, + r#""time":1781713834.069,"host":"web-01","index":"main","source":"vector-src","sourcetype":"vector_demo"}"#, + r#"{"event":{"message":"POST /login 401","status":"401"},"fields":{"region":"us-west"},"#, + r#""time":1781713834.119,"host":"web-02","index":"main","source":"vector-src","sourcetype":"vector_demo"}"#, + ); + let events = parse_hec_events(vector.as_bytes()).unwrap(); + assert_eq!(events.len(), 2); // concatenated, no separator + let (table, map, tags) = + hec_event_to_map(events.into_iter().next().unwrap(), None).unwrap(); + assert_eq!(table, "main"); + let tagset: HashSet<&str> = tags.iter().map(String::as_str).collect(); + assert_eq!( + tagset, + HashSet::from(["host", "source", "sourcetype", "region"]) + ); + let VrlValue::Object(m) = map else { + panic!("expected object"); + }; + assert!(matches!( + m.get(greptime_timestamp()), + Some(VrlValue::Timestamp(dt)) if dt.timestamp() == 1781713834 + )); + assert!(matches!(m.get("event"), Some(VrlValue::Object(_)))); // event is an object + + // --- OTel Collector splunk_hec exporter: NEWLINE-separated; `event` is a + // string; a `fields` key contains dots. --- + let otel = concat!( + r#"{"event":"{\"level\":\"info\",\"msg\":\"login ok\"}","fields":{"log.file.name":"app.log"},"#, + r#""host":"unknown","source":"otel-src","sourcetype":"otel_st","index":"main","time":1781714234.6849608}"#, + "\n", + r#"{"event":"{\"level\":\"error\",\"msg\":\"disk full\"}","fields":{"log.file.name":"app.log"},"#, + r#""host":"unknown","source":"otel-src","sourcetype":"otel_st","index":"main","time":1781714234.6849632}"#, + ); + let events = parse_hec_events(otel.as_bytes()).unwrap(); + assert_eq!(events.len(), 2); // newline-separated + let (table, map, tags) = + hec_event_to_map(events.into_iter().next().unwrap(), None).unwrap(); + assert_eq!(table, "main"); + let tagset: HashSet<&str> = tags.iter().map(String::as_str).collect(); + // a dotted `fields` key still becomes a tag column. + assert_eq!( + tagset, + HashSet::from(["host", "source", "sourcetype", "log.file.name"]) + ); + let VrlValue::Object(m) = map else { + panic!("expected object"); + }; + assert!(matches!( + m.get(greptime_timestamp()), + Some(VrlValue::Timestamp(dt)) if dt.timestamp() == 1781714234 + )); + assert!(matches!(m.get("event"), Some(VrlValue::Bytes(_)))); // event is a string + } + + #[test] + fn tag_promotion_is_scoped_per_table() { + use api::v1::{ColumnDataType, ColumnSchema, RowInsertRequest, Rows}; + + fn field_col(name: &str) -> ColumnSchema { + ColumnSchema { + column_name: name.to_string(), + datatype: ColumnDataType::String as i32, + semantic_type: SemanticType::Field as i32, + datatype_extension: None, + options: None, + } + } + fn req(table: &str, cols: &[&str]) -> RowInsertRequest { + RowInsertRequest { + table_name: table.to_string(), + rows: Some(Rows { + schema: cols.iter().map(|c| field_col(c)).collect(), + rows: vec![], + }), + } + } + + // One batch -> two tables, both with a `region` column. `region` is a tag in + // table "a" only; table "b"'s same-named field must NOT be promoted. + let ctx_req = + ContextReq::default_opt_with_reqs(vec![req("a", &["region"]), req("b", &["region"])]); + let mut tags: HashMap> = HashMap::new(); + tags.insert("a".to_string(), HashSet::from(["region".to_string()])); + tags.insert("b".to_string(), HashSet::new()); + + let out = apply_tag_columns(ctx_req, &tags); + + for r in out.ref_all_req() { + let region = r + .rows + .as_ref() + .unwrap() + .schema + .iter() + .find(|c| c.column_name == "region") + .unwrap(); + let expected = match r.table_name.as_str() { + "a" => SemanticType::Tag as i32, + "b" => SemanticType::Field as i32, // would have been Tag before the per-table fix + other => panic!("unexpected table {other}"), + }; + assert_eq!(region.semantic_type, expected, "table {}", r.table_name); + } + } +} diff --git a/src/session/src/context.rs b/src/session/src/context.rs index 864db9f1f1..3d22f812fd 100644 --- a/src/session/src/context.rs +++ b/src/session/src/context.rs @@ -608,6 +608,7 @@ pub enum Channel { Jaeger = 11, Log = 12, Promql = 13, + Splunk = 14, } impl From for Channel { @@ -626,6 +627,7 @@ impl From for Channel { 11 => Self::Jaeger, 12 => Self::Log, 13 => Self::Promql, + 14 => Self::Splunk, _ => Self::Unknown, } } @@ -663,6 +665,7 @@ impl AsRef for Channel { Channel::Jaeger => "jaeger", Channel::Log => "log", Channel::Promql => "promql", + Channel::Splunk => "splunk", Channel::Unknown => "unknown", } } diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index b37cb6ee81..fa46c3a0d2 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -157,6 +157,9 @@ macro_rules! http_tests { test_loki_json_logs_with_pipeline, test_elasticsearch_logs, test_elasticsearch_logs_with_index, + test_splunk_health, + test_splunk_health_is_public, + test_splunk_logs, test_log_query, test_jaeger_query_api, test_jaeger_query_api_for_trace_v1, @@ -1404,6 +1407,277 @@ pub async fn test_metrics_api(store_type: StorageType) { guard.remove_all().await; } +pub async fn test_splunk_health(store_type: StorageType) { + common_telemetry::init_default_ut_logging(); + let (app, mut guard) = + setup_test_http_app_with_frontend(store_type, "test_splunk_health").await; + let client = TestClient::new(app).await; + + // The HEC health endpoint returns 200 with the HEC health body. + let res = client + .get("/v1/splunk/services/collector/health") + .send() + .await; + assert_eq!(StatusCode::OK, res.status()); + let json: serde_json::Value = serde_json::from_str(&res.text().await).unwrap(); + assert_eq!(json["text"], "HEC is healthy"); + assert_eq!(json["code"], 17); + + // The versioned alias `/health/1.0` serves the same handler. + let res = client + .get("/v1/splunk/services/collector/health/1.0") + .send() + .await; + assert_eq!(StatusCode::OK, res.status()); + let json: serde_json::Value = serde_json::from_str(&res.text().await).unwrap(); + assert_eq!(json["text"], "HEC is healthy"); + assert_eq!(json["code"], 17); + + // Query parameters (e.g. `ack`, `token`) are tolerated and ignored rather + // than rejected + for path in [ + "/v1/splunk/services/collector/health?ack=true&token=xyz", + "/v1/splunk/services/collector/health/1.0?ack=true&token=xyz", + ] { + let res = client.get(path).send().await; + assert_eq!(StatusCode::OK, res.status()); + } + + guard.remove_all().await; +} + +pub async fn test_splunk_health_is_public(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, _guard) = setup_test_http_app_with_frontend_and_user_provider( + store_type, + "test_splunk_health_is_public", + Some(user_provider), + ) + .await; + let client = TestClient::new(app).await; + + // The health endpoint is registered in `PUBLIC_API_PREFIX`, so it must + // succeed without credentials even when a user provider is configured. + let res = client + .get("/v1/splunk/services/collector/health") + .send() + .await; + assert_eq!(StatusCode::OK, res.status()); +} + +pub async fn test_splunk_logs(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_logs", + Some(user_provider), + ) + .await; + let client = TestClient::new(app).await; + + // Authenticated SQL query (the user-provider harness requires auth on /v1/sql). + 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 ` + JSON content type. + let splunk_headers = || { + vec![ + ( + HeaderName::from_static("authorization"), + HeaderValue::from_static("Splunk greptime_user:greptime_pwd"), + ), + ( + HeaderName::from_static("content-type"), + HeaderValue::from_static("application/json"), + ), + ] + }; + let event_path = "/v1/splunk/services/collector/event"; + + // 1. Ingest a HEC batch (no `index` -> default `splunk_logs` table). + let body = concat!( + r#"{"event":"login ok","time":1700000000,"host":"web-01","source":"auth.log","sourcetype":"syslog","fields":{"region":"us-east"}}"#, + "\n", + r#"{"event":"login fail","time":1700000001,"host":"web-02","source":"auth.log","sourcetype":"syslog","fields":{"region":"us-west"}}"#, + ); + let res = send_req( + &client, + splunk_headers(), + event_path, + body.as_bytes().to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + assert!(res.text().await.contains("\"code\":0")); + + // 2. Rows landed with the right values (proves route + auth + parse + insert, + // and that the default table name is `splunk_logs`). + let rows = get_rows_from_output( + &query( + &client, + "select host, region, event, greptime_timestamp from splunk_logs order by host", + ) + .await, + ); + // `time` (epoch seconds) maps to the nanosecond timestamp column. + assert_eq!( + rows, + r#"[["web-01","us-east","login ok",1700000000000000000],["web-02","us-west","login fail",1700000001000000000]]"# + ); + + // 3. host/source/sourcetype + `fields` keys are tags (i.e. primary key). + let create = query(&client, "show create table splunk_logs").await; + let pk = create + .split("PRIMARY KEY") + .nth(1) + .expect("splunk_logs should have a PRIMARY KEY"); + for col in ["host", "source", "sourcetype", "region"] { + assert!( + pk.contains(col), + "expected `{col}` in primary key: {create}" + ); + } + // host/source/sourcetype lead the primary key (ahead of the `fields` tag `region`), + let pos = |col: &str| pk.find(col).expect("column missing from primary key"); + assert!( + pos("host") < pos("source") + && pos("source") < pos("sourcetype") + && pos("sourcetype") < pos("region"), + "expected host/source/sourcetype to lead the primary key: {create}" + ); + + // 4. A brand-new `fields` key on a later write also becomes a tag (dynamic + // tag column added to the existing table's primary key). + let body2 = r#"{"event":"deploy","time":1700000002,"host":"web-03","source":"deploy.log","sourcetype":"syslog","fields":{"datacenter":"dc1"}}"#; + let res = send_req( + &client, + splunk_headers(), + event_path, + body2.as_bytes().to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + + let create = query(&client, "show create table splunk_logs").await; + let pk = create.split("PRIMARY KEY").nth(1).unwrap(); + assert!( + pk.contains("datacenter"), + "expected dynamically-added `datacenter` in primary key: {create}" + ); + + // 5. An invalid `?table=` override returns HEC code 7 ("incorrect index"). + let res = send_req( + &client, + splunk_headers(), + "/v1/splunk/services/collector/event?table=bad%20name", + br#"{"event":"x","time":1700000003}"#.to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::BAD_REQUEST, res.status()); + assert!(res.text().await.contains("\"code\":7")); + + // 6. A custom pipeline (via `x-greptime-pipeline-name`) overrides identity and + // owns the schema: this one keeps host/event/timestamp and drops source/sourcetype. + let pipeline_yaml = r#" +transform: + - field: host + type: string + index: tag + - field: event + type: string + - field: greptime_timestamp + type: time + index: timestamp +"#; + let res = client + .post("/v1/pipelines/splunk_custom") + .header("Content-Type", "application/x-yaml") + .header("Authorization", basic_auth("greptime_user", "greptime_pwd")) + .body(pipeline_yaml) + .send() + .await; + assert_eq!(res.status(), StatusCode::OK, "create pipeline failed"); + + let mut headers = splunk_headers(); + headers.push(( + HeaderName::from_static("x-greptime-pipeline-name"), + HeaderValue::from_static("splunk_custom"), + )); + let res = send_req( + &client, + headers, + "/v1/splunk/services/collector/event?table=splunk_custom_tbl", + br#"{"event":"hi","time":1700000010,"host":"web-09","source":"s","sourcetype":"st"}"# + .to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::OK, res.status()); + + let create = query(&client, "show create table splunk_custom_tbl").await; + assert!( + create.contains("host"), + "custom pipeline kept host: {create}" + ); + assert!( + !create.contains("sourcetype"), + "custom pipeline should have dropped sourcetype (identity would keep it): {create}" + ); + + // 7. Auth failures return HEC codes: missing token -> 2 (401), bad token -> 4 (403). + let res = send_req( + &client, + vec![( + HeaderName::from_static("content-type"), + HeaderValue::from_static("application/json"), + )], + event_path, + br#"{"event":"x","time":1700000020}"#.to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::UNAUTHORIZED, res.status()); + assert!(res.text().await.contains("\"code\":2")); + + let res = send_req( + &client, + vec![ + ( + HeaderName::from_static("authorization"), + HeaderValue::from_static("Splunk baduser:badpass"), + ), + ( + HeaderName::from_static("content-type"), + HeaderValue::from_static("application/json"), + ), + ], + event_path, + br#"{"event":"x","time":1700000021}"#.to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::FORBIDDEN, res.status()); + assert!(res.text().await.contains("\"code\":4")); + + 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;