featherbit/plugins/native/
udp_logger.rs

1//! The `udp-logger` node — ships per-request access-log entries to a remote
2//! UDP endpoint, mirroring APISIX's `udp-logger` plugin.
3//!
4//! Entries are built with the shared [`build_entry`](crate::plugins::util::log_entry)
5//! helper, buffered in a [`BatchSink`], and delivered by a background task that
6//! binds an ephemeral [`tokio::net::UdpSocket`] and sends each entry as one
7//! datagram of JSON bytes to `host:port`. UDP is inherently fire-and-forget: a
8//! send that the kernel accepts is considered delivered; only local errors
9//! (name resolution, socket) surface as flush failures.
10//!
11//! Delivery never blocks the request path — [`Plugin::execute`] pushes the
12//! entry and returns the context unchanged — so this node should be placed in
13//! the response pipeline **after the `upstream` node**.
14//!
15//! ## Deviations from APISIX
16//! - Each entry is sent as its own datagram (one JSON object per packet),
17//!   rather than APISIX's "single object when `batch_max_size == 1`, JSON array
18//!   otherwise" shape.
19
20use std::collections::HashMap;
21use std::sync::Arc;
22use std::time::Duration;
23
24use async_trait::async_trait;
25use serde_json::Value;
26use tokio::net::UdpSocket;
27
28use crate::batch::{BatchConfig, BatchFlusher, BatchSink, FlushError};
29use crate::context::Context;
30use crate::plugins::util::log_entry::{build_entry, parse_log_format};
31use crate::plugins::{Plugin, PluginOutput, PluginResult};
32
33/// Serializes one entry to its datagram payload (compact JSON bytes). Pure and
34/// network-independent for unit testing.
35fn entry_to_datagram(entry: &Value) -> Vec<u8> {
36    serde_json::to_vec(entry).unwrap_or_else(|_| b"null".to_vec())
37}
38
39/// [`BatchFlusher`] that sends each entry as one UDP datagram to `host:port`.
40struct UdpFlusher {
41    host: String,
42    port: u16,
43    timeout: Duration,
44}
45
46impl UdpFlusher {
47    async fn send_all(&self, entries: &[Value]) -> Result<usize, (usize, String)> {
48        let addr = format!("{}:{}", self.host, self.port);
49        let socket = UdpSocket::bind("0.0.0.0:0")
50            .await
51            .map_err(|e| (0usize, format!("failed to bind UDP socket: {e}")))?;
52        for (i, entry) in entries.iter().enumerate() {
53            let payload = entry_to_datagram(entry);
54            let send = socket.send_to(&payload, &addr);
55            match tokio::time::timeout(self.timeout, send).await {
56                Ok(Ok(_)) => {}
57                Ok(Err(e)) => return Err((i, format!("failed to send to UDP server {addr}: {e}"))),
58                Err(_) => return Err((i, format!("timed out sending to UDP server {addr}"))),
59            }
60        }
61        Ok(entries.len())
62    }
63}
64
65#[async_trait]
66impl BatchFlusher for UdpFlusher {
67    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
68        match self.send_all(entries).await {
69            Ok(_) => Ok(()),
70            // Entries before `i` were already sent; retry only the tail.
71            Err((i, message)) => Err(FlushError {
72                message,
73                first_fail: Some(i),
74            }),
75        }
76    }
77}
78
79/// The `udp-logger` plugin node.
80pub struct UdpLoggerPlugin {
81    sink: BatchSink,
82    log_format: Option<HashMap<String, Value>>,
83    include_req_body: bool,
84    include_resp_body: bool,
85}
86
87impl UdpLoggerPlugin {
88    /// Builds the plugin from node config.
89    ///
90    /// Config keys:
91    /// - `host` (string, **required**): UDP server hostname or IP.
92    /// - `port` (integer, **required**): UDP server port.
93    /// - `timeout` (integer seconds, default `3`): per-datagram send timeout.
94    /// - `log_format` (object): custom `name -> "$var"` entry.
95    /// - `include_req_body` / `include_resp_body` (bool, default `false`).
96    /// - batch keys — see [`BatchConfig::from_config`].
97    ///
98    /// ```yaml
99    /// - id: udp-log
100    ///   type: udp-logger
101    ///   config:
102    ///     host: 127.0.0.1
103    ///     port: 5140
104    /// ```
105    pub fn from_config(config: &HashMap<String, Value>) -> Result<Self, String> {
106        let host = config
107            .get("host")
108            .and_then(|v| v.as_str())
109            .filter(|s| !s.is_empty())
110            .ok_or("udp-logger: `host` is required")?
111            .to_string();
112        let port = config
113            .get("port")
114            .and_then(|v| v.as_u64())
115            .filter(|p| *p <= u16::MAX as u64)
116            .ok_or("udp-logger: `port` is required and must be 0-65535")? as u16;
117
118        // APISIX udp-logger `timeout` is in seconds.
119        let timeout_s = config.get("timeout").and_then(|v| v.as_u64()).unwrap_or(3);
120        let timeout = Duration::from_secs(timeout_s.max(1));
121
122        let log_format = parse_log_format(config)?;
123        let include_req_body = config
124            .get("include_req_body")
125            .and_then(|v| v.as_bool())
126            .unwrap_or(false);
127        let include_resp_body = config
128            .get("include_resp_body")
129            .and_then(|v| v.as_bool())
130            .unwrap_or(false);
131
132        let batch_cfg = BatchConfig::from_config(config)?;
133        let flusher = Arc::new(UdpFlusher {
134            host: host.clone(),
135            port,
136            timeout,
137        });
138        let sink = BatchSink::spawn(&format!("udp-logger:{host}:{port}"), batch_cfg, flusher);
139
140        Ok(Self {
141            sink,
142            log_format,
143            include_req_body,
144            include_resp_body,
145        })
146    }
147}
148
149#[async_trait]
150impl Plugin for UdpLoggerPlugin {
151    fn plugin_type(&self) -> &str {
152        "udp-logger"
153    }
154
155    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
156        let entry = build_entry(
157            &ctx,
158            self.log_format.as_ref(),
159            self.include_req_body,
160            self.include_resp_body,
161        );
162        self.sink.push(entry);
163        Ok(PluginOutput {
164            context: ctx,
165            named_outputs: HashMap::new(),
166        })
167    }
168}
169
170#[cfg(test)]
171mod tests {
172    use super::*;
173    use serde_json::json;
174
175    fn base_config() -> HashMap<String, Value> {
176        let mut c = HashMap::new();
177        c.insert("host".to_string(), json!("127.0.0.1"));
178        c.insert("port".to_string(), json!(5140));
179        c
180    }
181
182    #[test]
183    fn from_config_requires_host_and_port() {
184        assert!(UdpLoggerPlugin::from_config(&HashMap::new()).is_err());
185        let mut c = HashMap::new();
186        c.insert("port".to_string(), json!(514));
187        assert!(
188            UdpLoggerPlugin::from_config(&c).is_err(),
189            "missing host must fail"
190        );
191    }
192
193    #[tokio::test]
194    async fn from_config_ok_with_valid_config() {
195        assert!(UdpLoggerPlugin::from_config(&base_config()).is_ok());
196    }
197
198    #[test]
199    fn entry_to_datagram_is_compact_json() {
200        let payload = entry_to_datagram(&json!({"a": 1, "b": "x"}));
201        let back: Value = serde_json::from_slice(&payload).unwrap();
202        assert_eq!(back, json!({"a": 1, "b": "x"}));
203    }
204}
featherbit/plugins/native/udp_logger.rs

featherbit/plugins/native/
udp_logger.rs
