feat: Add support for splunk HEC endpoints (#8321)

* feat: add health endpoint for splunk HEC

Signed-off-by: yash <yasha4658@gmail.com>

* feat: add event endpoint and handler + tests

Signed-off-by: yash <yasha4658@gmail.com>

* feat: add pipeline override test

Signed-off-by: yash <yasha4658@gmail.com>

* feat: add error response for auth errors

Signed-off-by: yash <yasha4658@gmail.com>

* fix: cargo fmt + doc example test

Signed-off-by: yash <yasha4658@gmail.com>

* fix: clippy warnings

Signed-off-by: yash <yasha4658@gmail.com>

* fix: address comments

Signed-off-by: yash <yasha4658@gmail.com>

* fix: address copilot comments and default table name inlining

Signed-off-by: yash <yasha4658@gmail.com>

* fix: cargo fmt

Signed-off-by: yash <yasha4658@gmail.com>

* fix: add error code 12, 13 and fix the timestamp override bug

Signed-off-by: yash <yasha4658@gmail.com>

* fix: add host/source/sourcetype ordering

Signed-off-by: yash <yasha4658@gmail.com>

* fix: fix typos

Signed-off-by: yash <yasha4658@gmail.com>

---------

Signed-off-by: yash <yasha4658@gmail.com>

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.

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<S>(log_state: LogState) -> Router<S> {
+        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<S>(log_state: LogState) -> Router<S> {
         Router::new()
             // Return fake responsefor HEAD '/' request.

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<B>(
             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<B>(
             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<B>(request: &Request<B>) -> Result<Option<(Username,
     }
 }
 
+fn get_splunk_credentials<B>(request: &Request<B>) -> Result<Option<(Username, Password)>> {
+    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<B>(request: &Request<B>) -> 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>` -> (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");

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<Item = Result<VrlValue, serde_json::Error>>,
     ignore_error: bool,
 ) -> Result<Vec<VrlValue>> {

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<Vec<VrlValue>> {
+    let values = Deserializer::from_slice(body).into_iter::<VrlValue>();
+    // 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<DateTime<Utc>> {
+    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<String>)> {
+    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<KeyString, VrlValue> = BTreeMap::new();
+    map.insert(
+        KeyString::from(greptime_timestamp()),
+        VrlValue::Timestamp(ts),
+    );
+
+    let mut tag_names: Vec<String> = 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<String, HashSet<String>>,
+) -> ContextReq {
+    let mut reqs = ctx_req.all_req().collect::<Vec<_>>();
+    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<String> {
+    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<B>(req: &axum::extract::Request<B>) -> bool {
+    // Match only `/v1/splunk/<subpath>`
+    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<PipelineIngestRequest>,
+    query_ctx: QueryContextRef,
+    pipeline_params: GreptimePipelineParams,
+    tag_columns: HashMap<String, HashSet<String>>,
+    apply_tags: bool,
+) -> Result<HttpResponse> {
+    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<LogState>,
+    Query(params): Query<LogIngesterQueryParams>,
+    Extension(mut query_ctx): Extension<QueryContext>,
+    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<String, Vec<VrlValue>> = HashMap::new();
+    let mut tag_columns: HashMap<String, HashSet<String>> = 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<PipelineIngestRequest> = 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<VrlValue> {
+        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(&not_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<String, HashSet<String>> = 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);
+        }
+    }
+}

src/session/src/context.rs

@@ -608,6 +608,7 @@ pub enum Channel {
     Jaeger = 11,
     Log = 12,
     Promql = 13,
+    Splunk = 14,
 }
 
 impl From<u32> for Channel {
@@ -626,6 +627,7 @@ impl From<u32> for Channel {
             11 => Self::Jaeger,
             12 => Self::Log,
             13 => Self::Promql,
+            14 => Self::Splunk,
             _ => Self::Unknown,
         }
     }
@@ -663,6 +665,7 @@ impl AsRef<str> for Channel {
             Channel::Jaeger => "jaeger",
             Channel::Log => "log",
             Channel::Promql => "promql",
+            Channel::Splunk => "splunk",
             Channel::Unknown => "unknown",
         }
     }

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 <user:pass>` + 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;