featherbit/plugins/native/
lago.rs

1//! The `lago` node — meters API traffic into [Lago](https://getlago.com), an
2//! open-source usage-based billing platform.
3//!
4//! Ported from APISIX's `lago.lua`. Unlike the other loggers this node is a
5//! **metering** integration, not an access log: it emits **one billing event
6//! per request** so Lago can price API consumption against a customer
7//! subscription. Each request produces a Lago usage event; events are buffered
8//! by a [`BatchSink`] and POSTed as a batch to `<endpoint>/api/v1/events/batch`
9//! with a `Bearer <token>` header. The node passes the context through
10//! unchanged.
11//!
12//! ## Deviations from APISIX
13//! - APISIX takes `endpoint_addrs` (an array, one picked at random per flush).
14//!   featherbit takes a single `endpoint`.
15//! - APISIX builds `properties` from a configured `event_properties` map of
16//!   `$var` templates. featherbit reuses the shared log-entry builder, so
17//!   `properties` is the standard request/response log entry (or a
18//!   `log_format` custom entry). This keeps the metering payload consistent
19//!   with the other loggers; configure `log_format` to shape it.
20
21use std::collections::HashMap;
22use std::sync::Arc;
23use std::time::Duration;
24
25use async_trait::async_trait;
26use serde_json::{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};
34use crate::vars;
35
36/// Emits one Lago billing event per request, delivered in batches.
37pub struct LagoPlugin {
38    sink: BatchSink,
39    log_format: Option<HashMap<String, Value>>,
40    include_req_body: bool,
41    include_resp_body: bool,
42    event_transaction_id: String,
43    subscription_id: String,
44    event_code: String,
45}
46
47/// Delivers batched usage events to `<endpoint>/api/v1/events/batch`.
48struct LagoFlusher {
49    client: Arc<OutboundClient>,
50    url: String,
51    token: String,
52    timeout: Duration,
53    ssl_verify: bool,
54}
55
56impl LagoPlugin {
57    /// Builds the plugin from node config.
58    ///
59    /// Config keys:
60    ///
61    /// | Key | Type | Default | Description |
62    /// |---|---|---|---|
63    /// | `endpoint` | string | — (required) | Lago API base, e.g. `http://127.0.0.1:3000`. |
64    /// | `token` | string | — (required) | Lago API key, sent as `Authorization: Bearer <token>`. |
65    /// | `event_transaction_id` | string | — (required) | `$var` template for the event's idempotency/transaction id, e.g. `req_$request_uri`. |
66    /// | `subscription_id` | string | — (required) | `$var` template identifying the customer subscription, e.g. `cus_$consumer_name`. |
67    /// | `event_code` | string | — (required) | Lago billable-metric code the event bills against. |
68    /// | `endpoint_uri` | string | `/api/v1/events/batch` | Batch-send path appended to `endpoint`. |
69    /// | `ssl_verify` | bool | `true` | Verify the Lago TLS certificate. |
70    /// | `timeout` | int (ms) | `3000` | Per-flush HTTP timeout. |
71    /// | `log_format` | object | — | Custom `name -> "$var template"` entry used for the event `properties`. |
72    /// | `include_req_body` / `include_resp_body` | bool | `false` | Include bodies in the default `properties` entry. |
73    ///
74    /// Batch keys default `batch_max_size` to **100** (Lago's batch limit)
75    /// rather than 1000; other batch keys follow [`BatchConfig::from_config`].
76    ///
77    /// ```yaml
78    /// type: lago
79    /// config:
80    ///   endpoint: http://127.0.0.1:3000
81    ///   token: ${LAGO_API_KEY}
82    ///   event_transaction_id: req_$request_uri
83    ///   subscription_id: cus_$consumer_name
84    ///   event_code: api_calls
85    /// ```
86    pub fn from_config(
87        config: &HashMap<String, Value>,
88        resources: &Arc<PluginResources>,
89    ) -> Result<Self, String> {
90        let endpoint = required_str(config, "endpoint")?
91            .trim_end_matches('/')
92            .to_string();
93        let token = required_str(config, "token")?.to_string();
94        let event_transaction_id = required_str(config, "event_transaction_id")?.to_string();
95        let subscription_id = required_str(config, "subscription_id")?.to_string();
96        let event_code = required_str(config, "event_code")?.to_string();
97
98        let endpoint_uri = config
99            .get("endpoint_uri")
100            .and_then(|v| v.as_str())
101            .unwrap_or("/api/v1/events/batch");
102        let ssl_verify = config
103            .get("ssl_verify")
104            .and_then(|v| v.as_bool())
105            .unwrap_or(true);
106        let timeout = Duration::from_millis(
107            config
108                .get("timeout")
109                .and_then(|v| v.as_u64())
110                .unwrap_or(3000),
111        );
112
113        let log_format = parse_log_format(config)?;
114        let include_req_body = config
115            .get("include_req_body")
116            .and_then(|v| v.as_bool())
117            .unwrap_or(false);
118        let include_resp_body = config
119            .get("include_resp_body")
120            .and_then(|v| v.as_bool())
121            .unwrap_or(false);
122
123        let mut batch_cfg = BatchConfig::from_config(config).map_err(|e| format!("lago: {e}"))?;
124        // Lago's batch endpoint caps a batch at 100 events; default to that.
125        if !config.contains_key("batch_max_size") {
126            batch_cfg.batch_max_size = 100;
127        }
128
129        let flusher = Arc::new(LagoFlusher {
130            client: resources.outbound.clone(),
131            url: format!("{endpoint}{endpoint_uri}"),
132            token,
133            timeout,
134            ssl_verify,
135        });
136        let sink = BatchSink::spawn("lago", batch_cfg, flusher);
137
138        Ok(Self {
139            sink,
140            log_format,
141            include_req_body,
142            include_resp_body,
143            event_transaction_id,
144            subscription_id,
145            event_code,
146        })
147    }
148}
149
150/// Builds one Lago usage event.
151fn build_event(
152    transaction_id: &str,
153    external_subscription_id: &str,
154    code: &str,
155    timestamp: u64,
156    properties: Value,
157) -> Value {
158    json!({
159        "transaction_id": transaction_id,
160        "external_subscription_id": external_subscription_id,
161        "code": code,
162        "timestamp": timestamp,
163        "properties": properties,
164    })
165}
166
167/// Wraps a batch of events into the Lago batch-events request body.
168fn build_batch_body(events: &[Value]) -> Value {
169    json!({ "events": events })
170}
171
172fn required_str<'a>(config: &'a HashMap<String, Value>, key: &str) -> Result<&'a str, String> {
173    config
174        .get(key)
175        .and_then(|v| v.as_str())
176        .filter(|s| !s.is_empty())
177        .ok_or_else(|| format!("lago: `{key}` is required"))
178}
179
180fn now_secs() -> u64 {
181    std::time::SystemTime::now()
182        .duration_since(std::time::UNIX_EPOCH)
183        .map(|d| d.as_secs())
184        .unwrap_or(0)
185}
186
187#[async_trait]
188impl BatchFlusher for LagoFlusher {
189    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
190        let body = serde_json::to_vec(&build_batch_body(entries)).map_err(|e| FlushError {
191            message: format!("failed to encode events batch: {e}"),
192            first_fail: None,
193        })?;
194
195        let req = OutboundRequest {
196            method: http::Method::POST,
197            url: self.url.clone(),
198            headers: vec![
199                ("Content-Type".to_string(), "application/json".to_string()),
200                (
201                    "Authorization".to_string(),
202                    format!("Bearer {}", self.token),
203                ),
204            ],
205            body: body.into(),
206            timeout: self.timeout,
207            ssl_verify: self.ssl_verify,
208            tls: None,
209        };
210
211        match self.client.request(req).await {
212            Ok(resp) if resp.status < 300 => Ok(()),
213            Ok(resp) => Err(FlushError {
214                message: format!(
215                    "lago api returned status {}: {}",
216                    resp.status,
217                    String::from_utf8_lossy(&resp.body)
218                ),
219                first_fail: None,
220            }),
221            Err(e) => Err(FlushError {
222                message: e.to_string(),
223                first_fail: None,
224            }),
225        }
226    }
227}
228
229#[async_trait]
230impl Plugin for LagoPlugin {
231    fn plugin_type(&self) -> &str {
232        "lago"
233    }
234
235    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
236        let transaction_id = vars::interpolate(&ctx, &self.event_transaction_id);
237        let external_subscription_id = vars::interpolate(&ctx, &self.subscription_id);
238        let properties = build_entry(
239            &ctx,
240            self.log_format.as_ref(),
241            self.include_req_body,
242            self.include_resp_body,
243        );
244        let event = build_event(
245            &transaction_id,
246            &external_subscription_id,
247            &self.event_code,
248            now_secs(),
249            properties,
250        );
251        self.sink.push(event);
252
253        Ok(PluginOutput {
254            context: ctx,
255            named_outputs: HashMap::new(),
256        })
257    }
258}
259
260#[cfg(test)]
261mod tests {
262    use super::*;
263
264    fn cfg(pairs: &[(&str, Value)]) -> HashMap<String, Value> {
265        pairs
266            .iter()
267            .map(|(k, v)| (k.to_string(), v.clone()))
268            .collect()
269    }
270
271    fn full_cfg() -> HashMap<String, Value> {
272        cfg(&[
273            ("endpoint", json!("http://lago:3000/")),
274            ("token", json!("secret")),
275            ("event_transaction_id", json!("req_$request_uri")),
276            ("subscription_id", json!("cus_$consumer_name")),
277            ("event_code", json!("api_calls")),
278        ])
279    }
280
281    #[test]
282    fn from_config_requires_fields() {
283        for missing in [
284            "endpoint",
285            "token",
286            "event_transaction_id",
287            "subscription_id",
288            "event_code",
289        ] {
290            let mut c = full_cfg();
291            c.remove(missing);
292            let res = LagoPlugin::from_config(&c, &PluginResources::empty());
293            let Err(e) = res else {
294                panic!("expected error when `{missing}` missing")
295            };
296            assert!(e.contains(missing));
297        }
298    }
299
300    #[tokio::test]
301    async fn from_config_ok() {
302        let p = LagoPlugin::from_config(&full_cfg(), &PluginResources::empty()).unwrap();
303        assert_eq!(p.event_code, "api_calls");
304        assert_eq!(p.event_transaction_id, "req_$request_uri");
305    }
306
307    #[test]
308    fn event_envelope_shape() {
309        let props = json!({ "request": { "method": "GET" } });
310        let ev = build_event("txn-1", "sub-9", "api_calls", 1_700_000_000, props);
311        assert_eq!(ev["transaction_id"], "txn-1");
312        assert_eq!(ev["external_subscription_id"], "sub-9");
313        assert_eq!(ev["code"], "api_calls");
314        assert_eq!(ev["timestamp"], 1_700_000_000u64);
315        assert_eq!(ev["properties"]["request"]["method"], "GET");
316    }
317
318    #[test]
319    fn batch_body_wraps_events() {
320        let events = vec![
321            json!({ "transaction_id": "a" }),
322            json!({ "transaction_id": "b" }),
323        ];
324        let body = build_batch_body(&events);
325        assert_eq!(body["events"].as_array().unwrap().len(), 2);
326        assert_eq!(body["events"][1]["transaction_id"], "b");
327    }
328}
featherbit/plugins/native/lago.rs

featherbit/plugins/native/
lago.rs
