featherbit/plugins/native/
splunk_hec_logging.rs

1//! The `splunk-hec-logging` node — ships access logs to a Splunk HTTP Event
2//! Collector (HEC) in batches.
3//!
4//! Ports the APISIX `splunk-hec-logging` plugin onto featherbit's shared
5//! [`BatchSink`](crate::batch::BatchSink). Each request builds a log entry,
6//! wraps it in a Splunk HEC event envelope (`{time, source, sourcetype,
7//! event}`), and hands it to the sink with a non-blocking `push`; a background
8//! task POSTs the concatenated events to the HEC `uri` with an
9//! `Authorization: Splunk <token>` header. The node never mutates the
10//! request/response and never fails, so it belongs in the response pipeline,
11//! **after the upstream node**.
12
13use std::collections::HashMap;
14use std::sync::Arc;
15use std::time::{Duration, SystemTime, UNIX_EPOCH};
16
17use async_trait::async_trait;
18use bytes::Bytes;
19use serde_json::{json, Value};
20
21use crate::batch::{BatchConfig, BatchFlusher, BatchSink, FlushError};
22use crate::context::Context;
23use crate::outbound::{OutboundClient, OutboundRequest};
24use crate::plugins::resources::PluginResources;
25use crate::plugins::util::log_entry::{build_entry, parse_log_format};
26use crate::plugins::{Plugin, PluginOutput, PluginResult};
27
28const DEFAULT_SOURCE: &str = "featherbit-splunk-hec-logging";
29const DEFAULT_SOURCETYPE: &str = "_json";
30
31/// Wraps one log entry in a Splunk HEC event envelope with a shared timestamp.
32fn wrap_event(entry: &Value, time: f64, source: &str) -> Value {
33    json!({
34        "time": time,
35        "source": source,
36        "sourcetype": DEFAULT_SOURCETYPE,
37        "event": entry,
38    })
39}
40
41/// Serializes a batch into the HEC request body: HEC accepts multiple JSON
42/// event objects concatenated with no separator. Factored out for unit testing.
43fn build_splunk_body(entries: &[Value], time: f64, source: &str) -> Bytes {
44    let mut buf = String::new();
45    for e in entries {
46        let event = wrap_event(e, time, source);
47        buf.push_str(&serde_json::to_string(&event).unwrap_or_default());
48    }
49    Bytes::from(buf)
50}
51
52/// Delivers batches by POSTing HEC events to the Splunk endpoint.
53struct SplunkFlusher {
54    client: Arc<OutboundClient>,
55    uri: String,
56    token: String,
57    channel: Option<String>,
58    source: String,
59    ssl_verify: bool,
60    timeout: Duration,
61}
62
63#[async_trait]
64impl BatchFlusher for SplunkFlusher {
65    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
66        let now = SystemTime::now()
67            .duration_since(UNIX_EPOCH)
68            .map(|d| d.as_secs_f64())
69            .unwrap_or(0.0);
70        let body = build_splunk_body(entries, now, &self.source);
71
72        let mut headers = vec![
73            ("Content-Type".to_string(), "application/json".to_string()),
74            (
75                "Authorization".to_string(),
76                format!("Splunk {}", self.token),
77            ),
78        ];
79        if let Some(ch) = &self.channel {
80            headers.push(("X-Splunk-Request-Channel".to_string(), ch.clone()));
81        }
82
83        let req = OutboundRequest {
84            method: http::Method::POST,
85            url: self.uri.clone(),
86            headers,
87            body,
88            timeout: self.timeout,
89            ssl_verify: self.ssl_verify,
90            tls: None,
91        };
92        match self.client.request(req).await {
93            Ok(resp) if resp.status == 200 => Ok(()),
94            Ok(resp) => Err(FlushError {
95                message: format!("splunk HEC returned status {}", resp.status),
96                first_fail: None,
97            }),
98            Err(e) => Err(FlushError {
99                message: e.to_string(),
100                first_fail: None,
101            }),
102        }
103    }
104}
105
106/// Batches access-log entries and POSTs them to a Splunk HEC endpoint.
107pub struct SplunkHecLoggingPlugin {
108    sink: BatchSink,
109    log_format: Option<HashMap<String, Value>>,
110    include_req_body: bool,
111    include_resp_body: bool,
112}
113
114impl SplunkHecLoggingPlugin {
115    /// Builds the plugin from node config.
116    ///
117    /// Accepted keys (the APISIX `endpoint` sub-object is flattened; both
118    /// `endpoint.uri` and a top-level `uri` are accepted):
119    /// - `endpoint.uri` (string, **required**): the HEC collector URL, e.g.
120    ///   `https://splunk:8088/services/collector`.
121    /// - `endpoint.token` (string, **required**): HEC token, sent as
122    ///   `Authorization: Splunk <token>`.
123    /// - `endpoint.channel` (string, optional): sent as the
124    ///   `X-Splunk-Request-Channel` header.
125    /// - `endpoint.timeout` (integer seconds, default `10`): whole-call
126    ///   deadline per flush.
127    /// - `source` (string, default `featherbit-splunk-hec-logging`): HEC event
128    ///   `source` field.
129    /// - `ssl_verify` (bool, default `true`): verify TLS certificates.
130    /// - `log_format`, `include_req_body`, `include_resp_body` — see the shared
131    ///   log-entry builder.
132    /// - Batch keys — see [`BatchConfig`].
133    ///
134    /// ```yaml
135    /// type: splunk-hec-logging
136    /// config:
137    ///   endpoint:
138    ///     uri: https://splunk:8088/services/collector
139    ///     token: 00000000-0000-0000-0000-000000000000
140    ///   ssl_verify: true
141    ///   batch_max_size: 1000
142    /// ```
143    pub fn from_config(
144        config: &HashMap<String, Value>,
145        resources: &Arc<PluginResources>,
146    ) -> Result<Self, String> {
147        let endpoint = config.get("endpoint").and_then(|v| v.as_object());
148
149        let get_ep = |key: &str| -> Option<&Value> {
150            endpoint
151                .and_then(|e| e.get(key))
152                .or_else(|| config.get(key))
153        };
154
155        let uri = get_ep("uri")
156            .and_then(|v| v.as_str())
157            .filter(|s| !s.is_empty())
158            .ok_or_else(|| "splunk-hec-logging plugin requires 'endpoint.uri'".to_string())?
159            .to_string();
160        let token = get_ep("token")
161            .and_then(|v| v.as_str())
162            .filter(|s| !s.is_empty())
163            .ok_or_else(|| "splunk-hec-logging plugin requires 'endpoint.token'".to_string())?
164            .to_string();
165        let channel = get_ep("channel")
166            .and_then(|v| v.as_str())
167            .filter(|s| !s.is_empty())
168            .map(|s| s.to_string());
169        let timeout = Duration::from_secs(
170            get_ep("timeout")
171                .and_then(|v| v.as_u64())
172                .unwrap_or(10)
173                .max(1),
174        );
175
176        let source = config
177            .get("source")
178            .and_then(|v| v.as_str())
179            .filter(|s| !s.is_empty())
180            .unwrap_or(DEFAULT_SOURCE)
181            .to_string();
182
183        let ssl_verify = config
184            .get("ssl_verify")
185            .and_then(|v| v.as_bool())
186            .unwrap_or(true);
187
188        let log_format = parse_log_format(config)?;
189        let include_req_body = config
190            .get("include_req_body")
191            .and_then(|v| v.as_bool())
192            .unwrap_or(false);
193        let include_resp_body = config
194            .get("include_resp_body")
195            .and_then(|v| v.as_bool())
196            .unwrap_or(false);
197
198        let batch_cfg = BatchConfig::from_config(config)?;
199        let flusher = Arc::new(SplunkFlusher {
200            client: resources.outbound.clone(),
201            uri,
202            token,
203            channel,
204            source,
205            ssl_verify,
206            timeout,
207        });
208        let sink = BatchSink::spawn("splunk-hec-logging", batch_cfg, flusher);
209
210        Ok(Self {
211            sink,
212            log_format,
213            include_req_body,
214            include_resp_body,
215        })
216    }
217}
218
219#[async_trait]
220impl Plugin for SplunkHecLoggingPlugin {
221    fn plugin_type(&self) -> &str {
222        "splunk-hec-logging"
223    }
224
225    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
226        let entry = build_entry(
227            &ctx,
228            self.log_format.as_ref(),
229            self.include_req_body,
230            self.include_resp_body,
231        );
232        self.sink.push(entry);
233        Ok(PluginOutput {
234            context: ctx,
235            named_outputs: HashMap::new(),
236        })
237    }
238}
239
240#[cfg(test)]
241mod tests {
242    use super::*;
243    use serde_json::json;
244
245    fn cfg(v: Value) -> HashMap<String, Value> {
246        serde_json::from_value(v).unwrap()
247    }
248
249    #[test]
250    fn rejects_missing_uri_or_token() {
251        // no endpoint at all
252        assert!(
253            SplunkHecLoggingPlugin::from_config(&HashMap::new(), &PluginResources::empty())
254                .is_err()
255        );
256        // uri without token
257        let c = cfg(json!({ "endpoint": { "uri": "https://splunk:8088/x" } }));
258        assert!(SplunkHecLoggingPlugin::from_config(&c, &PluginResources::empty()).is_err());
259    }
260
261    #[tokio::test]
262    async fn accepts_uri_and_token() {
263        let c = cfg(json!({ "endpoint": { "uri": "https://splunk:8088/x", "token": "tok" } }));
264        assert!(SplunkHecLoggingPlugin::from_config(&c, &PluginResources::empty()).is_ok());
265    }
266
267    #[test]
268    fn event_envelope_wraps_entry() {
269        let entry = json!({ "request": { "method": "GET" } });
270        let ev = wrap_event(&entry, 1.5, "src");
271        assert_eq!(ev["source"], "src");
272        assert_eq!(ev["sourcetype"], "_json");
273        assert_eq!(ev["time"], 1.5);
274        assert_eq!(ev["event"]["request"]["method"], "GET");
275    }
276
277    #[test]
278    fn body_concatenates_events_without_separator() {
279        let entries = vec![json!({"a": 1}), json!({"a": 2})];
280        let body = build_splunk_body(&entries, 0.0, DEFAULT_SOURCE);
281        let text = String::from_utf8(body.to_vec()).unwrap();
282        // Two JSON objects back-to-back: "}{"  boundary, no comma/newline.
283        assert!(text.contains("}{"));
284        assert!(!text.contains("},{"));
285        assert_eq!(text.matches("\"event\"").count(), 2);
286    }
287}
featherbit/plugins/native/splunk_hec_logging.rs

featherbit/plugins/native/
splunk_hec_logging.rs
