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, LogFormat};
33use crate::plugins::{Plugin, PluginOutput, PluginResult};
34use crate::vars::template::Template;
35
36/// Emits one Lago billing event per request, delivered in batches.
37pub struct LagoPlugin {
38    sink: BatchSink,
39    log_format: Option<LogFormat>,
40    include_req_body: bool,
41    include_resp_body: bool,
42    /// Template for the event's transaction id: supports
43    /// `{{namespace.path}}` references and legacy `$var` interpolation (see
44    /// [`Template::render_with_legacy`]).
45    event_transaction_id: Template,
46    /// Template for the subscription id: same rendering as
47    /// `event_transaction_id`.
48    subscription_id: Template,
49    event_code: String,
50}
51
52/// Delivers batched usage events to `<endpoint>/api/v1/events/batch`.
53struct LagoFlusher {
54    client: Arc<OutboundClient>,
55    url: String,
56    token: String,
57    timeout: Duration,
58    ssl_verify: bool,
59}
60
61impl LagoPlugin {
62    /// Builds the plugin from node config.
63    ///
64    /// Config keys:
65    ///
66    /// | Key | Type | Default | Description |
67    /// |---|---|---|---|
68    /// | `endpoint` | string | — (required) | Lago API base, e.g. `http://127.0.0.1:3000`. |
69    /// | `token` | string | — (required) | Lago API key, sent as `Authorization: Bearer <token>`. |
70    /// | `event_transaction_id` | string | — (required) | Template (`{{namespace.path}}` or legacy `$var`) for the event's idempotency/transaction id, e.g. `req_$request_uri`. |
71    /// | `subscription_id` | string | — (required) | Template (`{{namespace.path}}` or legacy `$var`) identifying the customer subscription, e.g. `cus_$consumer_name`. |
72    /// | `event_code` | string | — (required) | Lago billable-metric code the event bills against. |
73    /// | `endpoint_uri` | string | `/api/v1/events/batch` | Batch-send path appended to `endpoint`. |
74    /// | `ssl_verify` | bool | `true` | Verify the Lago TLS certificate. |
75    /// | `timeout` | int (ms) | `3000` | Per-flush HTTP timeout. |
76    /// | `log_format` | object | — | Custom `name -> "template"` entry (`{{namespace.path}}` references plus legacy `$var` interpolation) used for the event `properties`. |
77    /// | `include_req_body` / `include_resp_body` | bool | `false` | Include bodies in the default `properties` entry. |
78    ///
79    /// Batch keys default `batch_max_size` to **100** (Lago's batch limit)
80    /// rather than 1000; other batch keys follow [`BatchConfig::from_config`].
81    ///
82    /// ```yaml
83    /// type: lago
84    /// config:
85    ///   endpoint: http://127.0.0.1:3000
86    ///   token: ${LAGO_API_KEY}
87    ///   event_transaction_id: req_$request_uri
88    ///   subscription_id: cus_$consumer_name
89    ///   event_code: api_calls
90    /// ```
91    pub fn from_config(
92        config: &HashMap<String, Value>,
93        resources: &Arc<PluginResources>,
94    ) -> Result<Self, String> {
95        let endpoint = required_str(config, "endpoint")?
96            .trim_end_matches('/')
97            .to_string();
98        let token = required_str(config, "token")?.to_string();
99        // Discard warnings here — the compile-time walk (a later task)
100        // reports well-formed-but-unknown references; execution must not.
101        let event_transaction_id = Template::parse(required_str(config, "event_transaction_id")?).0;
102        let subscription_id = Template::parse(required_str(config, "subscription_id")?).0;
103        let event_code = required_str(config, "event_code")?.to_string();
104
105        let endpoint_uri = config
106            .get("endpoint_uri")
107            .and_then(|v| v.as_str())
108            .unwrap_or("/api/v1/events/batch");
109        let ssl_verify = config
110            .get("ssl_verify")
111            .and_then(|v| v.as_bool())
112            .unwrap_or(true);
113        let timeout = Duration::from_millis(
114            config
115                .get("timeout")
116                .and_then(|v| v.as_u64())
117                .unwrap_or(3000),
118        );
119
120        let log_format = parse_log_format(config)?;
121        let include_req_body = config
122            .get("include_req_body")
123            .and_then(|v| v.as_bool())
124            .unwrap_or(false);
125        let include_resp_body = config
126            .get("include_resp_body")
127            .and_then(|v| v.as_bool())
128            .unwrap_or(false);
129
130        let mut batch_cfg = BatchConfig::from_config(config).map_err(|e| format!("lago: {e}"))?;
131        // Lago's batch endpoint caps a batch at 100 events; default to that.
132        if !config.contains_key("batch_max_size") {
133            batch_cfg.batch_max_size = 100;
134        }
135
136        let flusher = Arc::new(LagoFlusher {
137            client: resources.outbound.clone(),
138            url: format!("{endpoint}{endpoint_uri}"),
139            token,
140            timeout,
141            ssl_verify,
142        });
143        let sink = BatchSink::spawn("lago", batch_cfg, flusher);
144
145        Ok(Self {
146            sink,
147            log_format,
148            include_req_body,
149            include_resp_body,
150            event_transaction_id,
151            subscription_id,
152            event_code,
153        })
154    }
155}
156
157/// Builds one Lago usage event.
158fn build_event(
159    transaction_id: &str,
160    external_subscription_id: &str,
161    code: &str,
162    timestamp: u64,
163    properties: Value,
164) -> Value {
165    json!({
166        "transaction_id": transaction_id,
167        "external_subscription_id": external_subscription_id,
168        "code": code,
169        "timestamp": timestamp,
170        "properties": properties,
171    })
172}
173
174/// Wraps a batch of events into the Lago batch-events request body.
175fn build_batch_body(events: &[Value]) -> Value {
176    json!({ "events": events })
177}
178
179fn required_str<'a>(config: &'a HashMap<String, Value>, key: &str) -> Result<&'a str, String> {
180    config
181        .get(key)
182        .and_then(|v| v.as_str())
183        .filter(|s| !s.is_empty())
184        .ok_or_else(|| format!("lago: `{key}` is required"))
185}
186
187fn now_secs() -> u64 {
188    std::time::SystemTime::now()
189        .duration_since(std::time::UNIX_EPOCH)
190        .map(|d| d.as_secs())
191        .unwrap_or(0)
192}
193
194#[async_trait]
195impl BatchFlusher for LagoFlusher {
196    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
197        let body = serde_json::to_vec(&build_batch_body(entries)).map_err(|e| FlushError {
198            message: format!("failed to encode events batch: {e}"),
199            first_fail: None,
200        })?;
201
202        let req = OutboundRequest {
203            method: http::Method::POST,
204            url: self.url.clone(),
205            headers: vec![
206                ("Content-Type".to_string(), "application/json".to_string()),
207                (
208                    "Authorization".to_string(),
209                    format!("Bearer {}", self.token),
210                ),
211            ],
212            body: body.into(),
213            timeout: self.timeout,
214            ssl_verify: self.ssl_verify,
215            tls: None,
216        };
217
218        match self.client.request(req).await {
219            Ok(resp) if resp.status < 300 => Ok(()),
220            Ok(resp) => Err(FlushError {
221                message: format!(
222                    "lago api returned status {}: {}",
223                    resp.status,
224                    String::from_utf8_lossy(&resp.body)
225                ),
226                first_fail: None,
227            }),
228            Err(e) => Err(FlushError {
229                message: e.to_string(),
230                first_fail: None,
231            }),
232        }
233    }
234}
235
236#[async_trait]
237impl Plugin for LagoPlugin {
238    fn plugin_type(&self) -> &str {
239        "lago"
240    }
241
242    fn reads_response_body(&self) -> bool {
243        crate::plugins::util::log_entry::reads_response_body(
244            self.log_format.as_ref(),
245            self.include_resp_body,
246        )
247    }
248
249    async fn execute(&self, ctx: Context) -> PluginResult {
250        let transaction_id = self.event_transaction_id.render_with_legacy(&ctx);
251        let external_subscription_id = self.subscription_id.render_with_legacy(&ctx);
252        let properties = build_entry(
253            &ctx,
254            self.log_format.as_ref(),
255            self.include_req_body,
256            self.include_resp_body,
257        );
258        let event = build_event(
259            &transaction_id,
260            &external_subscription_id,
261            &self.event_code,
262            now_secs(),
263            properties,
264        );
265        self.sink.push(event);
266
267        Ok(PluginOutput::success(ctx))
268    }
269}
270
271#[cfg(test)]
272mod tests {
273    use super::*;
274
275    fn cfg(pairs: &[(&str, Value)]) -> HashMap<String, Value> {
276        pairs
277            .iter()
278            .map(|(k, v)| (k.to_string(), v.clone()))
279            .collect()
280    }
281
282    fn full_cfg() -> HashMap<String, Value> {
283        cfg(&[
284            ("endpoint", json!("http://lago:3000/")),
285            ("token", json!("secret")),
286            ("event_transaction_id", json!("req_$request_uri")),
287            ("subscription_id", json!("cus_$consumer_name")),
288            ("event_code", json!("api_calls")),
289        ])
290    }
291
292    #[test]
293    fn from_config_requires_fields() {
294        for missing in [
295            "endpoint",
296            "token",
297            "event_transaction_id",
298            "subscription_id",
299            "event_code",
300        ] {
301            let mut c = full_cfg();
302            c.remove(missing);
303            let res = LagoPlugin::from_config(&c, &PluginResources::empty());
304            let Err(e) = res else {
305                panic!("expected error when `{missing}` missing")
306            };
307            assert!(e.contains(missing));
308        }
309    }
310
311    #[tokio::test]
312    async fn from_config_ok() {
313        let p = LagoPlugin::from_config(&full_cfg(), &PluginResources::empty()).unwrap();
314        assert_eq!(p.event_code, "api_calls");
315        // event_transaction_id is pre-parsed into a Template at construction;
316        // verify it still carries the configured legacy `$var` template by
317        // rendering it against a request whose path makes the substitution
318        // observable.
319        let ctx = crate::context::Context {
320            request: crate::context::GatewayRequest {
321                method: "GET".to_string(),
322                path: "/foo".to_string(),
323                host: "example.com".to_string(),
324                scheme: "http".to_string(),
325                headers: HashMap::new(),
326                query_params: HashMap::new(),
327                body: bytes::Bytes::new(),
328                remote_addr: "127.0.0.1:1".to_string(),
329                protocol: crate::context::Protocol::Http1,
330            },
331            response: crate::context::GatewayResponse {
332                status_code: 0,
333                headers: HashMap::new(),
334                body: bytes::Bytes::new(),
335                stream: None,
336            },
337            message: HashMap::new(),
338            errors: Vec::new(),
339        };
340        assert_eq!(p.event_transaction_id.render_with_legacy(&ctx), "req_/foo");
341    }
342
343    #[test]
344    fn event_envelope_shape() {
345        let props = json!({ "request": { "method": "GET" } });
346        let ev = build_event("txn-1", "sub-9", "api_calls", 1_700_000_000, props);
347        assert_eq!(ev["transaction_id"], "txn-1");
348        assert_eq!(ev["external_subscription_id"], "sub-9");
349        assert_eq!(ev["code"], "api_calls");
350        assert_eq!(ev["timestamp"], 1_700_000_000u64);
351        assert_eq!(ev["properties"]["request"]["method"], "GET");
352    }
353
354    #[test]
355    fn batch_body_wraps_events() {
356        let events = vec![
357            json!({ "transaction_id": "a" }),
358            json!({ "transaction_id": "b" }),
359        ];
360        let body = build_batch_body(&events);
361        assert_eq!(body["events"].as_array().unwrap().len(), 2);
362        assert_eq!(body["events"][1]["transaction_id"], "b");
363    }
364}
featherbit/plugins/native/lago.rs

featherbit/plugins/native/
lago.rs
