featherbit/plugins/native/
http_logger.rs

1//! The `http-logger` node — ships access logs to an arbitrary HTTP endpoint
2//! in batches.
3//!
4//! Ports the APISIX `http-logger` plugin onto featherbit's shared
5//! [`BatchSink`](crate::batch::BatchSink) and pooled outbound HTTP client.
6//! Each request builds a log entry via
7//! [`build_entry`](crate::plugins::util::log_entry::build_entry) and hands it
8//! to the sink with a non-blocking, fire-and-forget `push`; a background task
9//! POSTs accumulated batches to `uri`. The node never mutates the
10//! request/response and never fails, so it belongs in the response pipeline,
11//! **after the upstream node**, where the final status/body are available.
12
13use std::collections::HashMap;
14use std::sync::Arc;
15use std::time::Duration;
16
17use async_trait::async_trait;
18use bytes::Bytes;
19use serde_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
28/// How batched entries are serialized into the request body.
29#[derive(Debug, Clone, Copy, PartialEq, Eq)]
30enum ConcatMethod {
31    /// A single JSON array of entries (`application/json`).
32    Json,
33    /// One JSON object per line, `\n`-separated (`text/plain`).
34    NewLine,
35}
36
37/// Serializes a batch into an HTTP body plus its `Content-Type`. Factored out
38/// of the network path so payload shaping is unit-testable without a socket.
39fn build_http_body(entries: &[Value], concat: ConcatMethod) -> (Bytes, &'static str) {
40    match concat {
41        ConcatMethod::Json => {
42            let body = serde_json::to_vec(entries).unwrap_or_default();
43            (Bytes::from(body), "application/json")
44        }
45        ConcatMethod::NewLine => {
46            let lines: Vec<String> = entries
47                .iter()
48                .map(|e| serde_json::to_string(e).unwrap_or_default())
49                .collect();
50            (Bytes::from(lines.join("\n")), "text/plain")
51        }
52    }
53}
54
55/// Delivers batches by POSTing them to the configured HTTP endpoint.
56struct HttpLoggerFlusher {
57    client: Arc<OutboundClient>,
58    uri: String,
59    method: http::Method,
60    /// Extra headers applied to every request (e.g. `Authorization`).
61    headers: Vec<(String, String)>,
62    concat: ConcatMethod,
63    ssl_verify: bool,
64    timeout: Duration,
65}
66
67#[async_trait]
68impl BatchFlusher for HttpLoggerFlusher {
69    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
70        let (body, content_type) = build_http_body(entries, self.concat);
71        let mut headers = self.headers.clone();
72        headers.push(("Content-Type".to_string(), content_type.to_string()));
73
74        let req = OutboundRequest {
75            method: self.method.clone(),
76            url: self.uri.clone(),
77            headers,
78            body,
79            timeout: self.timeout,
80            ssl_verify: self.ssl_verify,
81            tls: None,
82        };
83        match self.client.request(req).await {
84            Ok(resp) if resp.status < 400 => Ok(()),
85            Ok(resp) => Err(FlushError {
86                message: format!("http-logger endpoint returned status {}", resp.status),
87                first_fail: None,
88            }),
89            Err(e) => Err(FlushError {
90                message: e.to_string(),
91                first_fail: None,
92            }),
93        }
94    }
95}
96
97/// Batches access-log entries and POSTs them to an HTTP endpoint.
98pub struct HttpLoggerPlugin {
99    sink: BatchSink,
100    log_format: Option<HashMap<String, Value>>,
101    include_req_body: bool,
102    include_resp_body: bool,
103}
104
105impl HttpLoggerPlugin {
106    /// Builds the plugin from node config.
107    ///
108    /// Accepted keys:
109    /// - `uri` (string, **required**): the HTTP endpoint batches are POSTed to.
110    ///   A missing/empty `uri` is a config-load error.
111    /// - `method` (string, default `POST`): HTTP method for the callout.
112    /// - `headers` (object of string→string, optional): extra request headers
113    ///   applied to every batch (e.g. custom auth).
114    /// - `auth_header` (string, optional): convenience for a single
115    ///   `Authorization` header value.
116    /// - `concat_method` (string, default `json`): `json` sends a JSON array
117    ///   (`application/json`); `new_line` sends `\n`-separated JSON objects
118    ///   (`text/plain`).
119    /// - `ssl_verify` (bool, default `false`): verify TLS certificates for
120    ///   `https` endpoints.
121    /// - `timeout` (integer seconds, default `3`): whole-call deadline per flush.
122    /// - `log_format` (object, optional): custom flat entry of
123    ///   `name -> "$var template"`; when absent the default structured entry is
124    ///   used.
125    /// - `include_req_body` / `include_resp_body` (bool, default `false`): add
126    ///   the request/response body to the default entry.
127    /// - Batch keys (`batch_max_size`, `inactive_timeout`, `buffer_duration`,
128    ///   `max_retry_count`, `retry_delay`, `max_pending_entries`) — see
129    ///   [`BatchConfig`].
130    ///
131    /// ```yaml
132    /// type: http-logger
133    /// config:
134    ///   uri: http://log-collector:3000/logs
135    ///   method: POST
136    ///   headers:
137    ///     X-Token: secret
138    ///   concat_method: json
139    ///   batch_max_size: 1000
140    ///   inactive_timeout: 5
141    /// ```
142    pub fn from_config(
143        config: &HashMap<String, Value>,
144        resources: &Arc<PluginResources>,
145    ) -> Result<Self, String> {
146        let uri = config
147            .get("uri")
148            .and_then(|v| v.as_str())
149            .filter(|s| !s.is_empty())
150            .ok_or_else(|| "http-logger plugin requires 'uri'".to_string())?
151            .to_string();
152
153        let method_str = config
154            .get("method")
155            .and_then(|v| v.as_str())
156            .unwrap_or("POST")
157            .to_uppercase();
158        let method = http::Method::from_bytes(method_str.as_bytes())
159            .map_err(|_| format!("http-logger invalid method '{}'", method_str))?;
160
161        let concat = match config
162            .get("concat_method")
163            .and_then(|v| v.as_str())
164            .unwrap_or("json")
165        {
166            "json" => ConcatMethod::Json,
167            "new_line" => ConcatMethod::NewLine,
168            other => {
169                return Err(format!(
170                    "http-logger concat_method must be 'json' or 'new_line', got '{}'",
171                    other
172                ))
173            }
174        };
175
176        let mut headers: Vec<(String, String)> = Vec::new();
177        if let Some(obj) = config.get("headers").and_then(|v| v.as_object()) {
178            for (k, v) in obj {
179                if let Some(s) = header_value(v) {
180                    headers.push((k.clone(), s));
181                }
182            }
183        }
184        if let Some(auth) = config.get("auth_header").and_then(|v| v.as_str()) {
185            if !auth.is_empty() {
186                headers.push(("Authorization".to_string(), auth.to_string()));
187            }
188        }
189
190        let ssl_verify = config
191            .get("ssl_verify")
192            .and_then(|v| v.as_bool())
193            .unwrap_or(false);
194        let timeout = Duration::from_secs(
195            config
196                .get("timeout")
197                .and_then(|v| v.as_u64())
198                .unwrap_or(3)
199                .max(1),
200        );
201
202        let log_format = parse_log_format(config)?;
203        let include_req_body = config
204            .get("include_req_body")
205            .and_then(|v| v.as_bool())
206            .unwrap_or(false);
207        let include_resp_body = config
208            .get("include_resp_body")
209            .and_then(|v| v.as_bool())
210            .unwrap_or(false);
211
212        let batch_cfg = BatchConfig::from_config(config)?;
213        let flusher = Arc::new(HttpLoggerFlusher {
214            client: resources.outbound.clone(),
215            uri,
216            method,
217            headers,
218            concat,
219            ssl_verify,
220            timeout,
221        });
222        let sink = BatchSink::spawn("http-logger", batch_cfg, flusher);
223
224        Ok(Self {
225            sink,
226            log_format,
227            include_req_body,
228            include_resp_body,
229        })
230    }
231}
232
233/// Coerces a JSON scalar to a header string; `None` for non-scalars.
234fn header_value(v: &Value) -> Option<String> {
235    match v {
236        Value::String(s) => Some(s.clone()),
237        Value::Number(n) => Some(n.to_string()),
238        Value::Bool(b) => Some(b.to_string()),
239        _ => None,
240    }
241}
242
243#[async_trait]
244impl Plugin for HttpLoggerPlugin {
245    fn plugin_type(&self) -> &str {
246        "http-logger"
247    }
248
249    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
250        let entry = build_entry(
251            &ctx,
252            self.log_format.as_ref(),
253            self.include_req_body,
254            self.include_resp_body,
255        );
256        self.sink.push(entry);
257        Ok(PluginOutput {
258            context: ctx,
259            named_outputs: HashMap::new(),
260        })
261    }
262}
263
264#[cfg(test)]
265mod tests {
266    use super::*;
267    use serde_json::json;
268
269    fn cfg(v: Value) -> HashMap<String, Value> {
270        serde_json::from_value(v).unwrap()
271    }
272
273    #[test]
274    fn rejects_missing_uri() {
275        assert!(HttpLoggerPlugin::from_config(&HashMap::new(), &PluginResources::empty()).is_err());
276        let c = cfg(json!({ "uri": "" }));
277        assert!(HttpLoggerPlugin::from_config(&c, &PluginResources::empty()).is_err());
278    }
279
280    #[tokio::test]
281    async fn accepts_minimal_config() {
282        let c = cfg(json!({ "uri": "http://collector/logs" }));
283        assert!(HttpLoggerPlugin::from_config(&c, &PluginResources::empty()).is_ok());
284    }
285
286    #[test]
287    fn rejects_bad_concat_method() {
288        let c = cfg(json!({ "uri": "http://c", "concat_method": "csv" }));
289        assert!(HttpLoggerPlugin::from_config(&c, &PluginResources::empty()).is_err());
290    }
291
292    #[test]
293    fn json_body_is_an_array() {
294        let entries = vec![json!({"a": 1}), json!({"a": 2})];
295        let (body, ct) = build_http_body(&entries, ConcatMethod::Json);
296        assert_eq!(ct, "application/json");
297        let parsed: Value = serde_json::from_slice(&body).unwrap();
298        assert!(parsed.is_array());
299        assert_eq!(parsed.as_array().unwrap().len(), 2);
300        assert_eq!(parsed[1]["a"], 2);
301    }
302
303    #[test]
304    fn new_line_body_is_newline_delimited() {
305        let entries = vec![json!({"a": 1}), json!({"a": 2})];
306        let (body, ct) = build_http_body(&entries, ConcatMethod::NewLine);
307        assert_eq!(ct, "text/plain");
308        let text = String::from_utf8(body.to_vec()).unwrap();
309        let lines: Vec<&str> = text.split('\n').collect();
310        assert_eq!(lines.len(), 2);
311        let first: Value = serde_json::from_str(lines[0]).unwrap();
312        assert_eq!(first["a"], 1);
313    }
314}
featherbit/plugins/native/http_logger.rs

featherbit/plugins/native/
http_logger.rs
