featherbit/plugins/native/
tcp_logger.rs

1//! The `tcp-logger` node — ships per-request access-log entries to a remote
2//! TCP endpoint (a log collector such as Logstash, Fluentd, or a raw TCP
3//! sink), mirroring APISIX's `tcp-logger` plugin.
4//!
5//! Entries are built with the shared [`build_entry`](crate::plugins::util::log_entry)
6//! helper, buffered in a [`BatchSink`], and delivered by a background task that
7//! opens a fresh [`tokio::net::TcpStream`] per flush, writes each entry as one
8//! newline-delimited JSON object, and closes the connection. Delivery is
9//! **fire-and-forget** on the request path: [`Plugin::execute`] never blocks on
10//! the network and always returns the context unchanged, so this node should be
11//! placed in the response pipeline **after the `upstream` node** where the
12//! final status and body size are available.
13//!
14//! ## Deviations from APISIX
15//! - Entries are always sent as newline-delimited JSON (one object per line),
16//!   rather than APISIX's "single object when `batch_max_size == 1`, JSON array
17//!   otherwise" shape. This keeps the wire format stable regardless of batching.
18//! - `tls` / `tls_options` are **not yet supported**; configuring `tls: true`
19//!   is rejected at config load. Plain TCP only.
20
21use std::collections::HashMap;
22use std::sync::Arc;
23use std::time::Duration;
24
25use async_trait::async_trait;
26use serde_json::Value;
27use tokio::io::AsyncWriteExt;
28use tokio::net::TcpStream;
29
30use crate::batch::{BatchConfig, BatchFlusher, BatchSink, FlushError};
31use crate::context::Context;
32use crate::plugins::util::log_entry::{build_entry, parse_log_format};
33use crate::plugins::{Plugin, PluginOutput, PluginResult};
34
35/// Serializes each entry as one line of JSON, newline-terminated, and
36/// concatenates them into the payload written to the socket. Pure and
37/// independent of the network so it can be unit-tested directly.
38fn entries_to_lines(entries: &[Value]) -> String {
39    let mut out = String::new();
40    for entry in entries {
41        // Value serialization is infallible; fall back to `null` defensively.
42        out.push_str(&serde_json::to_string(entry).unwrap_or_else(|_| "null".to_string()));
43        out.push('\n');
44    }
45    out
46}
47
48/// [`BatchFlusher`] that connects to `host:port` and writes the batch.
49struct TcpFlusher {
50    host: String,
51    port: u16,
52    timeout: Duration,
53}
54
55impl TcpFlusher {
56    async fn send(&self, payload: &[u8]) -> Result<(), String> {
57        let addr = format!("{}:{}", self.host, self.port);
58        let mut stream = tokio::time::timeout(self.timeout, TcpStream::connect(&addr))
59            .await
60            .map_err(|_| format!("timed out connecting to TCP server {addr}"))?
61            .map_err(|e| format!("failed to connect to TCP server {addr}: {e}"))?;
62        tokio::time::timeout(self.timeout, stream.write_all(payload))
63            .await
64            .map_err(|_| format!("timed out sending to TCP server {addr}"))?
65            .map_err(|e| format!("failed to send to TCP server {addr}: {e}"))?;
66        let _ = stream.shutdown().await;
67        Ok(())
68    }
69}
70
71#[async_trait]
72impl BatchFlusher for TcpFlusher {
73    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
74        let payload = entries_to_lines(entries);
75        self.send(payload.as_bytes())
76            .await
77            .map_err(|message| FlushError {
78                message,
79                // A TCP connect/write failure delivered nothing; retry the whole batch.
80                first_fail: None,
81            })
82    }
83}
84
85/// The `tcp-logger` plugin node.
86pub struct TcpLoggerPlugin {
87    sink: BatchSink,
88    log_format: Option<HashMap<String, Value>>,
89    include_req_body: bool,
90    include_resp_body: bool,
91}
92
93impl TcpLoggerPlugin {
94    /// Builds the plugin from node config.
95    ///
96    /// Config keys:
97    /// - `host` (string, **required**): TCP server hostname or IP.
98    /// - `port` (integer, **required**): TCP server port.
99    /// - `timeout` (integer ms, default `1000`): connect/send timeout.
100    /// - `tls` (bool, default `false`): **not yet supported** — `true` is rejected.
101    /// - `tls_options` (string): accepted but ignored (TLS unsupported).
102    /// - `log_format` (object): custom `name -> "$var"` entry; when set, the
103    ///   default structured entry is replaced by this flat object.
104    /// - `include_req_body` / `include_resp_body` (bool, default `false`): add
105    ///   the request/response body to the default entry.
106    /// - batch keys (`batch_max_size`, `inactive_timeout`, `buffer_duration`,
107    ///   `max_retry_count`, `retry_delay`, `max_pending_entries`) — see
108    ///   [`BatchConfig::from_config`].
109    ///
110    /// ```yaml
111    /// - id: tcp-log
112    ///   type: tcp-logger
113    ///   config:
114    ///     host: 127.0.0.1
115    ///     port: 5044
116    ///     timeout: 1000
117    ///     batch_max_size: 100
118    /// ```
119    pub fn from_config(config: &HashMap<String, Value>) -> Result<Self, String> {
120        let host = config
121            .get("host")
122            .and_then(|v| v.as_str())
123            .filter(|s| !s.is_empty())
124            .ok_or("tcp-logger: `host` is required")?
125            .to_string();
126        let port = config
127            .get("port")
128            .and_then(|v| v.as_u64())
129            .filter(|p| *p <= u16::MAX as u64)
130            .ok_or("tcp-logger: `port` is required and must be 0-65535")? as u16;
131
132        if config.get("tls").and_then(|v| v.as_bool()).unwrap_or(false) {
133            return Err("tcp-logger: TLS (`tls: true`) is not yet supported".to_string());
134        }
135
136        let timeout_ms = config
137            .get("timeout")
138            .and_then(|v| v.as_u64())
139            .unwrap_or(1000);
140        let timeout = Duration::from_millis(timeout_ms.max(1));
141
142        let log_format = parse_log_format(config)?;
143        let include_req_body = config
144            .get("include_req_body")
145            .and_then(|v| v.as_bool())
146            .unwrap_or(false);
147        let include_resp_body = config
148            .get("include_resp_body")
149            .and_then(|v| v.as_bool())
150            .unwrap_or(false);
151
152        let batch_cfg = BatchConfig::from_config(config)?;
153        let flusher = Arc::new(TcpFlusher {
154            host: host.clone(),
155            port,
156            timeout,
157        });
158        let sink = BatchSink::spawn(&format!("tcp-logger:{host}:{port}"), batch_cfg, flusher);
159
160        Ok(Self {
161            sink,
162            log_format,
163            include_req_body,
164            include_resp_body,
165        })
166    }
167}
168
169#[async_trait]
170impl Plugin for TcpLoggerPlugin {
171    fn plugin_type(&self) -> &str {
172        "tcp-logger"
173    }
174
175    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
176        let entry = build_entry(
177            &ctx,
178            self.log_format.as_ref(),
179            self.include_req_body,
180            self.include_resp_body,
181        );
182        self.sink.push(entry);
183        Ok(PluginOutput {
184            context: ctx,
185            named_outputs: HashMap::new(),
186        })
187    }
188}
189
190#[cfg(test)]
191mod tests {
192    use super::*;
193    use serde_json::json;
194
195    fn base_config() -> HashMap<String, Value> {
196        let mut c = HashMap::new();
197        c.insert("host".to_string(), json!("127.0.0.1"));
198        c.insert("port".to_string(), json!(5044));
199        c
200    }
201
202    #[test]
203    fn from_config_requires_host_and_port() {
204        assert!(TcpLoggerPlugin::from_config(&HashMap::new()).is_err());
205        let mut c = HashMap::new();
206        c.insert("host".to_string(), json!("h"));
207        assert!(
208            TcpLoggerPlugin::from_config(&c).is_err(),
209            "missing port must fail"
210        );
211    }
212
213    #[test]
214    fn from_config_rejects_tls() {
215        let mut c = base_config();
216        c.insert("tls".to_string(), json!(true));
217        let err = TcpLoggerPlugin::from_config(&c).err().unwrap();
218        assert!(err.contains("TLS"), "error should mention TLS: {err}");
219    }
220
221    #[tokio::test]
222    async fn from_config_ok_with_valid_config() {
223        // Needs a tokio runtime because BatchSink::spawn spawns a task.
224        assert!(TcpLoggerPlugin::from_config(&base_config()).is_ok());
225    }
226
227    #[test]
228    fn entries_to_lines_is_newline_delimited_json() {
229        let entries = vec![json!({"a": 1}), json!({"b": 2})];
230        let out = entries_to_lines(&entries);
231        assert_eq!(out, "{\"a\":1}\n{\"b\":2}\n");
232        // Each line round-trips as JSON.
233        for line in out.lines() {
234            let _: Value = serde_json::from_str(line).unwrap();
235        }
236    }
237}
featherbit/plugins/native/tcp_logger.rs

featherbit/plugins/native/
tcp_logger.rs
