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, LogFormat};
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<LogFormat>,
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    fn reads_response_body(&self) -> bool {
226        crate::plugins::util::log_entry::reads_response_body(
227            self.log_format.as_ref(),
228            self.include_resp_body,
229        )
230    }
231
232    async fn execute(&self, ctx: Context) -> PluginResult {
233        let entry = build_entry(
234            &ctx,
235            self.log_format.as_ref(),
236            self.include_req_body,
237            self.include_resp_body,
238        );
239        self.sink.push(entry);
240        Ok(PluginOutput::success(ctx))
241    }
242}
243
244#[cfg(test)]
245mod tests {
246    use super::*;
247    use serde_json::json;
248
249    fn cfg(v: Value) -> HashMap<String, Value> {
250        serde_json::from_value(v).unwrap()
251    }
252
253    #[test]
254    fn rejects_missing_uri_or_token() {
255        // no endpoint at all
256        assert!(
257            SplunkHecLoggingPlugin::from_config(&HashMap::new(), &PluginResources::empty())
258                .is_err()
259        );
260        // uri without token
261        let c = cfg(json!({ "endpoint": { "uri": "https://splunk:8088/x" } }));
262        assert!(SplunkHecLoggingPlugin::from_config(&c, &PluginResources::empty()).is_err());
263    }
264
265    #[tokio::test]
266    async fn accepts_uri_and_token() {
267        let c = cfg(json!({ "endpoint": { "uri": "https://splunk:8088/x", "token": "tok" } }));
268        assert!(SplunkHecLoggingPlugin::from_config(&c, &PluginResources::empty()).is_ok());
269    }
270
271    #[test]
272    fn event_envelope_wraps_entry() {
273        let entry = json!({ "request": { "method": "GET" } });
274        let ev = wrap_event(&entry, 1.5, "src");
275        assert_eq!(ev["source"], "src");
276        assert_eq!(ev["sourcetype"], "_json");
277        assert_eq!(ev["time"], 1.5);
278        assert_eq!(ev["event"]["request"]["method"], "GET");
279    }
280
281    #[test]
282    fn body_concatenates_events_without_separator() {
283        let entries = vec![json!({"a": 1}), json!({"a": 2})];
284        let body = build_splunk_body(&entries, 0.0, DEFAULT_SOURCE);
285        let text = String::from_utf8(body.to_vec()).unwrap();
286        // Two JSON objects back-to-back: "}{"  boundary, no comma/newline.
287        assert!(text.contains("}{"));
288        assert!(!text.contains("},{"));
289        assert_eq!(text.matches("\"event\"").count(), 2);
290    }
291}
featherbit/plugins/native/splunk_hec_logging.rs

featherbit/plugins/native/
splunk_hec_logging.rs
