featherbit/plugins/native/
datadog.rs

1//! The `datadog` node — ships request metrics to a Datadog agent over
2//! DogStatsD (UDP) in batches.
3//!
4//! **Deviation from the other loggers:** APISIX's `datadog` plugin does *not*
5//! ship JSON access logs to an HTTP endpoint. It emits DogStatsD metrics
6//! (`request.counter`, `request.latency`, `ingress.size`, `egress.size`, ...)
7//! as UDP datagrams to a local Datadog agent. This port stays faithful: the
8//! flusher is a [`tokio::net::UdpSocket`] rather than the shared outbound HTTP
9//! client, and each buffered log entry is rendered into DogStatsD metric lines.
10//! The APISIX `plugin-metadata` fields (`host`, `port`, `namespace`,
11//! `constant_tags`) are flattened into this node's config. Because featherbit's
12//! shared log entry carries no route/service identifiers, `prefer_name` is
13//! accepted for compatibility but has no effect, and no `route_name`/
14//! `service_name` tags are emitted.
15//!
16//! The node never mutates the request/response and never fails; place it in the
17//! response pipeline, **after the upstream node**.
18
19use std::collections::HashMap;
20use std::sync::Arc;
21
22use async_trait::async_trait;
23use serde_json::Value;
24use tokio::net::UdpSocket;
25
26use crate::batch::{BatchConfig, BatchFlusher, BatchSink, FlushError};
27use crate::context::Context;
28use crate::plugins::resources::PluginResources;
29use crate::plugins::util::log_entry::build_entry;
30use crate::plugins::{Plugin, PluginOutput, PluginResult};
31
32/// DogStatsD agent's default receive buffer; larger datagrams are truncated,
33/// so coalesced payloads are capped here (matching APISIX).
34const MAX_DATAGRAM_SIZE: usize = 8192;
35
36/// Builds the DogStatsD metric lines for one featherbit log entry.
37///
38/// Rendered as `namespace.metric:value|type|#tag,tag`. Factored out of the
39/// socket path so metric formatting is unit-testable.
40fn build_metric_lines(
41    entry: &Value,
42    namespace: &str,
43    constant_tags: &[String],
44    include_path: bool,
45    include_method: bool,
46) -> Vec<String> {
47    let prefix = if namespace.is_empty() {
48        String::new()
49    } else {
50        format!("{}.", namespace)
51    };
52
53    // Assemble tags.
54    let mut tags: Vec<String> = constant_tags.to_vec();
55    if include_method {
56        if let Some(m) = entry.pointer("/request/method").and_then(|v| v.as_str()) {
57            tags.push(format!("method:{}", m));
58        }
59    }
60    if include_path {
61        if let Some(p) = entry.pointer("/request/uri").and_then(|v| v.as_str()) {
62            tags.push(format!("path:{}", p));
63        }
64    }
65    if let Some(status) = entry.pointer("/response/status").and_then(|v| v.as_u64()) {
66        tags.push(format!("response_status:{}", status));
67        tags.push(format!("response_status_class:{}xx", status / 100));
68    }
69    if let Some(scheme) = entry.pointer("/request/scheme").and_then(|v| v.as_str()) {
70        tags.push(format!("scheme:{}", scheme));
71    }
72    let suffix = if tags.is_empty() {
73        String::new()
74    } else {
75        format!("|#{}", tags.join(","))
76    };
77
78    let mut lines = vec![format!("{}request.counter:1|c{}", prefix, suffix)];
79    if let Some(latency) = entry.get("latency").and_then(|v| v.as_u64()) {
80        lines.push(format!("{}request.latency:{}|h{}", prefix, latency, suffix));
81    }
82    if let Some(size) = entry.pointer("/request/size").and_then(|v| v.as_u64()) {
83        lines.push(format!("{}ingress.size:{}|ms{}", prefix, size, suffix));
84    }
85    if let Some(size) = entry.pointer("/response/size").and_then(|v| v.as_u64()) {
86        lines.push(format!("{}egress.size:{}|ms{}", prefix, size, suffix));
87    }
88    lines
89}
90
91/// Delivers batches by sending DogStatsD datagrams to the agent over UDP.
92struct DatadogFlusher {
93    addr: String,
94    namespace: String,
95    constant_tags: Vec<String>,
96    include_path: bool,
97    include_method: bool,
98}
99
100#[async_trait]
101impl BatchFlusher for DatadogFlusher {
102    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
103        let sock = UdpSocket::bind("0.0.0.0:0").await.map_err(|e| FlushError {
104            message: format!("udp bind failed: {}", e),
105            first_fail: None,
106        })?;
107        sock.connect(&self.addr).await.map_err(|e| FlushError {
108            message: format!("udp connect to {} failed: {}", self.addr, e),
109            first_fail: None,
110        })?;
111
112        for (i, entry) in entries.iter().enumerate() {
113            let lines = build_metric_lines(
114                entry,
115                &self.namespace,
116                &self.constant_tags,
117                self.include_path,
118                self.include_method,
119            );
120            let payload = lines.join("\n");
121            let send_res = if payload.len() <= MAX_DATAGRAM_SIZE {
122                sock.send(payload.as_bytes()).await.map(|_| ())
123            } else {
124                // Too large to coalesce: one datagram per metric line.
125                let mut r = Ok(());
126                for line in &lines {
127                    if let Err(e) = sock.send(line.as_bytes()).await {
128                        r = Err(e);
129                        break;
130                    }
131                }
132                r
133            };
134            if let Err(e) = send_res {
135                // Entries before `i` were already delivered; retry only the tail.
136                return Err(FlushError {
137                    message: format!("failed to send metrics to {}: {}", self.addr, e),
138                    first_fail: Some(i),
139                });
140            }
141        }
142        Ok(())
143    }
144}
145
146/// Batches request metrics and sends them to a Datadog agent via DogStatsD.
147pub struct DatadogPlugin {
148    sink: BatchSink,
149    include_path: bool,
150    include_method: bool,
151}
152
153impl DatadogPlugin {
154    /// Builds the plugin from node config.
155    ///
156    /// Accepted keys (APISIX `plugin-metadata` fields flattened in):
157    /// - `host` (string, default `127.0.0.1`): Datadog agent address.
158    /// - `port` (integer, default `8125`): DogStatsD UDP port.
159    /// - `namespace` (string, default `featherbit`): metric name prefix.
160    /// - `constant_tags` (array of strings, default `[]`): tags added to every
161    ///   metric.
162    /// - `include_path` (bool, default `false`): add a `path:` tag.
163    /// - `include_method` (bool, default `false`): add a `method:` tag.
164    /// - `prefer_name` (bool, default `true`): accepted for APISIX
165    ///   compatibility but has no effect here (no route/service names in the
166    ///   log entry).
167    /// - Batch keys — see [`BatchConfig`].
168    ///
169    /// ```yaml
170    /// type: datadog
171    /// config:
172    ///   host: 127.0.0.1
173    ///   port: 8125
174    ///   namespace: featherbit
175    ///   constant_tags: [source:featherbit]
176    ///   include_method: true
177    /// ```
178    pub fn from_config(
179        config: &HashMap<String, Value>,
180        _resources: &Arc<PluginResources>,
181    ) -> Result<Self, String> {
182        let host = config
183            .get("host")
184            .and_then(|v| v.as_str())
185            .unwrap_or("127.0.0.1");
186        let port = config.get("port").and_then(|v| v.as_u64()).unwrap_or(8125);
187        if port > 65535 {
188            return Err(format!("datadog port out of range: {}", port));
189        }
190        let addr = format!("{}:{}", host, port);
191
192        let namespace = config
193            .get("namespace")
194            .and_then(|v| v.as_str())
195            .unwrap_or("featherbit")
196            .to_string();
197
198        let constant_tags: Vec<String> = config
199            .get("constant_tags")
200            .and_then(|v| v.as_array())
201            .map(|arr| {
202                arr.iter()
203                    .filter_map(|v| v.as_str().map(String::from))
204                    .collect()
205            })
206            .unwrap_or_default();
207
208        let include_path = config
209            .get("include_path")
210            .and_then(|v| v.as_bool())
211            .unwrap_or(false);
212        let include_method = config
213            .get("include_method")
214            .and_then(|v| v.as_bool())
215            .unwrap_or(false);
216
217        // datadog batches at 5000 by default in APISIX; the shared BatchConfig
218        // default of 1000 is used unless overridden.
219        let batch_cfg = BatchConfig::from_config(config)?;
220        let flusher = Arc::new(DatadogFlusher {
221            addr,
222            namespace,
223            constant_tags,
224            include_path,
225            include_method,
226        });
227        let sink = BatchSink::spawn("datadog", batch_cfg, flusher);
228
229        Ok(Self {
230            sink,
231            include_path,
232            include_method,
233        })
234    }
235}
236
237#[async_trait]
238impl Plugin for DatadogPlugin {
239    fn plugin_type(&self) -> &str {
240        "datadog"
241    }
242
243    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
244        // Datadog derives metrics from the default structured entry; it does
245        // not support log_format or body capture.
246        let _ = (self.include_path, self.include_method);
247        let entry = build_entry(&ctx, None, false, false);
248        self.sink.push(entry);
249        Ok(PluginOutput {
250            context: ctx,
251            named_outputs: HashMap::new(),
252        })
253    }
254}
255
256#[cfg(test)]
257mod tests {
258    use super::*;
259    use serde_json::json;
260
261    fn cfg(v: Value) -> HashMap<String, Value> {
262        serde_json::from_value(v).unwrap()
263    }
264
265    fn entry() -> Value {
266        json!({
267            "request": { "method": "GET", "uri": "/api", "size": 12, "scheme": "https" },
268            "response": { "status": 200, "size": 34 },
269            "latency": 7
270        })
271    }
272
273    #[tokio::test]
274    async fn accepts_empty_config_with_defaults() {
275        let p = DatadogPlugin::from_config(&HashMap::new(), &PluginResources::empty());
276        assert!(p.is_ok());
277    }
278
279    #[test]
280    fn rejects_bad_port() {
281        let c = cfg(json!({ "port": 70000 }));
282        assert!(DatadogPlugin::from_config(&c, &PluginResources::empty()).is_err());
283    }
284
285    #[test]
286    fn metric_lines_cover_counter_latency_and_sizes() {
287        let lines = build_metric_lines(
288            &entry(),
289            "featherbit",
290            &["source:fb".to_string()],
291            true,
292            true,
293        );
294        let joined = lines.join("\n");
295        assert!(joined.contains("featherbit.request.counter:1|c"));
296        assert!(joined.contains("featherbit.request.latency:7|h"));
297        assert!(joined.contains("featherbit.ingress.size:12|ms"));
298        assert!(joined.contains("featherbit.egress.size:34|ms"));
299        // tags
300        assert!(joined.contains("source:fb"));
301        assert!(joined.contains("method:GET"));
302        assert!(joined.contains("path:/api"));
303        assert!(joined.contains("response_status:200"));
304        assert!(joined.contains("response_status_class:2xx"));
305        assert!(joined.contains("scheme:https"));
306    }
307
308    #[test]
309    fn empty_namespace_omits_prefix() {
310        let lines = build_metric_lines(&entry(), "", &[], false, false);
311        assert!(lines[0].starts_with("request.counter:1|c"));
312        // no method/path tags when disabled
313        assert!(!lines[0].contains("method:"));
314        assert!(!lines[0].contains("path:"));
315    }
316}
featherbit/plugins/native/datadog.rs

featherbit/plugins/native/
datadog.rs
