featherbit/plugins/native/
loggly.rs

1//! The `loggly` node — ships access logs to SolarWinds Loggly in batches via
2//! the HTTP/S bulk endpoint.
3//!
4//! Ports the APISIX `loggly` plugin onto featherbit's shared
5//! [`BatchSink`](crate::batch::BatchSink) and pooled outbound HTTP client.
6//! Each request builds a log entry and hands it to the sink with a
7//! non-blocking `push`; a background task POSTs the batch as newline-delimited
8//! JSON to `https://<host>/bulk/<customer_token>/tag/<tags>/`.
9//!
10//! **Deviation:** APISIX's `loggly` defaults to a syslog-over-UDP transport and
11//! only uses the HTTP bulk endpoint when its `plugin-metadata.protocol` is
12//! `http`/`https`. This port implements the HTTP/S bulk path only (the shape
13//! matching the other featherbit loggers); the RFC5424 syslog framing,
14//! `severity`/`severity_map`, and UDP transport are not implemented. `severity`
15//! is accepted for compatibility but ignored.
16//!
17//! The node never mutates the request/response and never fails; place it in the
18//! response pipeline, **after the upstream node**.
19
20use std::collections::HashMap;
21use std::sync::Arc;
22use std::time::Duration;
23
24use async_trait::async_trait;
25use bytes::Bytes;
26use serde_json::Value;
27
28use crate::batch::{BatchConfig, BatchFlusher, BatchSink, FlushError};
29use crate::context::Context;
30use crate::outbound::{OutboundClient, OutboundRequest};
31use crate::plugins::resources::PluginResources;
32use crate::plugins::util::log_entry::{build_entry, parse_log_format};
33use crate::plugins::{Plugin, PluginOutput, PluginResult};
34
35const DEFAULT_HOST: &str = "logs-01.loggly.com";
36
37/// Builds the Loggly bulk endpoint URL for a token and tag set.
38fn build_bulk_url(host: &str, token: &str, tags: &[String]) -> String {
39    let base = if host.starts_with("http://") || host.starts_with("https://") {
40        host.trim_end_matches('/').to_string()
41    } else {
42        format!("https://{}", host.trim_end_matches('/'))
43    };
44    let tag_seg = if tags.is_empty() {
45        "featherbit".to_string()
46    } else {
47        tags.join(",")
48    };
49    format!("{}/bulk/{}/tag/{}/", base, token, tag_seg)
50}
51
52/// Serializes a batch as newline-delimited JSON, the format the Loggly bulk
53/// endpoint expects. Factored out for unit testing.
54fn build_bulk_body(entries: &[Value]) -> Bytes {
55    let lines: Vec<String> = entries
56        .iter()
57        .map(|e| serde_json::to_string(e).unwrap_or_default())
58        .collect();
59    Bytes::from(lines.join("\n"))
60}
61
62/// Delivers batches by POSTing newline-delimited JSON to the Loggly bulk API.
63struct LogglyFlusher {
64    client: Arc<OutboundClient>,
65    url: String,
66    tags_header: String,
67    ssl_verify: bool,
68    timeout: Duration,
69}
70
71#[async_trait]
72impl BatchFlusher for LogglyFlusher {
73    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
74        let body = build_bulk_body(entries);
75        let headers = vec![
76            ("Content-Type".to_string(), "application/json".to_string()),
77            ("X-LOGGLY-TAG".to_string(), self.tags_header.clone()),
78        ];
79        let req = OutboundRequest {
80            method: http::Method::POST,
81            url: self.url.clone(),
82            headers,
83            body,
84            timeout: self.timeout,
85            ssl_verify: self.ssl_verify,
86            tls: None,
87        };
88        match self.client.request(req).await {
89            Ok(resp) if resp.status == 200 => Ok(()),
90            Ok(resp) => Err(FlushError {
91                message: format!("loggly returned status {}", resp.status),
92                first_fail: None,
93            }),
94            Err(e) => Err(FlushError {
95                message: e.to_string(),
96                first_fail: None,
97            }),
98        }
99    }
100}
101
102/// Batches access-log entries and POSTs them to the Loggly bulk endpoint.
103pub struct LogglyPlugin {
104    sink: BatchSink,
105    log_format: Option<HashMap<String, Value>>,
106    include_req_body: bool,
107    include_resp_body: bool,
108}
109
110impl LogglyPlugin {
111    /// Builds the plugin from node config.
112    ///
113    /// Accepted keys:
114    /// - `customer_token` (string, **required**): Loggly customer token; forms
115    ///   part of the bulk URL. Missing/empty is a config-load error.
116    /// - `tags` (array of strings, default `[featherbit]`): Loggly tags,
117    ///   comma-joined into the URL and the `X-LOGGLY-TAG` header.
118    /// - `host` (string, default `logs-01.loggly.com`): Loggly host; a bare
119    ///   host gets an `https://` scheme.
120    /// - `severity` (string, default `INFO`): accepted for APISIX
121    ///   compatibility but ignored (HTTP bulk mode carries no syslog severity).
122    /// - `ssl_verify` (bool, default `true`): verify TLS certificates.
123    /// - `timeout` (integer ms, default `5000`): whole-call deadline per flush.
124    /// - `log_format`, `include_req_body`, `include_resp_body` — see the shared
125    ///   log-entry builder.
126    /// - Batch keys — see [`BatchConfig`].
127    ///
128    /// ```yaml
129    /// type: loggly
130    /// config:
131    ///   customer_token: 00000000-0000-0000-0000-000000000000
132    ///   tags: [featherbit, prod]
133    ///   ssl_verify: true
134    /// ```
135    pub fn from_config(
136        config: &HashMap<String, Value>,
137        resources: &Arc<PluginResources>,
138    ) -> Result<Self, String> {
139        let token = config
140            .get("customer_token")
141            .and_then(|v| v.as_str())
142            .filter(|s| !s.is_empty())
143            .ok_or_else(|| "loggly plugin requires 'customer_token'".to_string())?
144            .to_string();
145
146        let tags: Vec<String> = config
147            .get("tags")
148            .and_then(|v| v.as_array())
149            .map(|arr| {
150                arr.iter()
151                    .filter_map(|v| v.as_str().map(String::from))
152                    .collect()
153            })
154            .unwrap_or_else(|| vec!["featherbit".to_string()]);
155
156        let host = config
157            .get("host")
158            .and_then(|v| v.as_str())
159            .unwrap_or(DEFAULT_HOST);
160        let url = build_bulk_url(host, &token, &tags);
161        let tags_header = if tags.is_empty() {
162            "featherbit".to_string()
163        } else {
164            tags.join(",")
165        };
166
167        let ssl_verify = config
168            .get("ssl_verify")
169            .and_then(|v| v.as_bool())
170            .unwrap_or(true);
171        let timeout = Duration::from_millis(
172            config
173                .get("timeout")
174                .and_then(|v| v.as_u64())
175                .unwrap_or(5000)
176                .max(1),
177        );
178
179        let log_format = parse_log_format(config)?;
180        let include_req_body = config
181            .get("include_req_body")
182            .and_then(|v| v.as_bool())
183            .unwrap_or(false);
184        let include_resp_body = config
185            .get("include_resp_body")
186            .and_then(|v| v.as_bool())
187            .unwrap_or(false);
188
189        let batch_cfg = BatchConfig::from_config(config)?;
190        let flusher = Arc::new(LogglyFlusher {
191            client: resources.outbound.clone(),
192            url,
193            tags_header,
194            ssl_verify,
195            timeout,
196        });
197        let sink = BatchSink::spawn("loggly", batch_cfg, flusher);
198
199        Ok(Self {
200            sink,
201            log_format,
202            include_req_body,
203            include_resp_body,
204        })
205    }
206}
207
208#[async_trait]
209impl Plugin for LogglyPlugin {
210    fn plugin_type(&self) -> &str {
211        "loggly"
212    }
213
214    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
215        let entry = build_entry(
216            &ctx,
217            self.log_format.as_ref(),
218            self.include_req_body,
219            self.include_resp_body,
220        );
221        self.sink.push(entry);
222        Ok(PluginOutput {
223            context: ctx,
224            named_outputs: HashMap::new(),
225        })
226    }
227}
228
229#[cfg(test)]
230mod tests {
231    use super::*;
232    use serde_json::json;
233
234    fn cfg(v: Value) -> HashMap<String, Value> {
235        serde_json::from_value(v).unwrap()
236    }
237
238    #[test]
239    fn rejects_missing_token() {
240        assert!(LogglyPlugin::from_config(&HashMap::new(), &PluginResources::empty()).is_err());
241        let c = cfg(json!({ "customer_token": "" }));
242        assert!(LogglyPlugin::from_config(&c, &PluginResources::empty()).is_err());
243    }
244
245    #[tokio::test]
246    async fn accepts_token() {
247        let c = cfg(json!({ "customer_token": "tok" }));
248        assert!(LogglyPlugin::from_config(&c, &PluginResources::empty()).is_ok());
249    }
250
251    #[test]
252    fn bulk_url_defaults_to_https_host() {
253        let url = build_bulk_url(DEFAULT_HOST, "tok", &["a".to_string(), "b".to_string()]);
254        assert_eq!(url, "https://logs-01.loggly.com/bulk/tok/tag/a,b/");
255    }
256
257    #[test]
258    fn bulk_url_respects_explicit_scheme() {
259        let url = build_bulk_url("http://loggly.local", "tok", &["x".to_string()]);
260        assert_eq!(url, "http://loggly.local/bulk/tok/tag/x/");
261    }
262
263    #[test]
264    fn bulk_body_is_newline_delimited_json() {
265        let entries = vec![json!({"a": 1}), json!({"a": 2})];
266        let body = build_bulk_body(&entries);
267        let text = String::from_utf8(body.to_vec()).unwrap();
268        let lines: Vec<&str> = text.split('\n').collect();
269        assert_eq!(lines.len(), 2);
270        let second: Value = serde_json::from_str(lines[1]).unwrap();
271        assert_eq!(second["a"], 2);
272    }
273}
featherbit/plugins/native/loggly.rs

featherbit/plugins/native/
loggly.rs
