featherbit/plugins/native/
error_log_logger.rs

1//! The `error-log-logger` node — a **reinterpreted** subset of APISIX's
2//! `error-log-logger` plugin.
3//!
4//! In APISIX this plugin tails Nginx's internal error-log stream (via
5//! `ngx.errlog`) and forwards those lines to a remote sink. featherbit has no
6//! separate internal error-log stream to tail, so the faithful reinterpretation
7//! is an **error-only access logger**: it builds the shared access-log entry and
8//! ships it to a remote TCP sink **only when the request accumulated errors**
9//! (`context.errors` is non-empty). Requests that succeed produce nothing.
10//!
11//! Entries are buffered in a [`BatchSink`] and delivered by a background task
12//! that opens a fresh [`tokio::net::TcpStream`] per flush and writes each entry
13//! as one newline-delimited JSON object. Delivery is fire-and-forget on the
14//! request path, so place this node in the response pipeline **after the
15//! `upstream` node** (and after any node whose errors you want captured).
16//!
17//! ## Deviations from APISIX
18//! - Source of logs: featherbit logs **request-level errors** (`context.errors`)
19//!   rather than the gateway's own internal error-log lines.
20//! - Only the `tcp` sink (`host`/`port`) is implemented; the `skywalking`,
21//!   `clickhouse`, and `kafka` sinks are out of scope for this socket-focused
22//!   node.
23//! - `level` is accepted for compatibility but featherbit filters on the
24//!   *presence* of request errors, not on a syslog severity threshold.
25//! - `tls` is **not yet supported**; `tls: true` is rejected at config load.
26
27use std::collections::HashMap;
28use std::sync::Arc;
29use std::time::Duration;
30
31use async_trait::async_trait;
32use serde_json::Value;
33use tokio::io::AsyncWriteExt;
34use tokio::net::TcpStream;
35
36use crate::batch::{BatchConfig, BatchFlusher, BatchSink, FlushError};
37use crate::context::Context;
38use crate::plugins::util::log_entry::{build_entry, parse_log_format};
39use crate::plugins::{Plugin, PluginOutput, PluginResult};
40
41/// Serializes each entry as one line of JSON, newline-terminated. Pure helper.
42fn entries_to_lines(entries: &[Value]) -> String {
43    let mut out = String::new();
44    for entry in entries {
45        out.push_str(&serde_json::to_string(entry).unwrap_or_else(|_| "null".to_string()));
46        out.push('\n');
47    }
48    out
49}
50
51/// [`BatchFlusher`] that connects to `host:port` and writes the batch.
52struct TcpFlusher {
53    host: String,
54    port: u16,
55    timeout: Duration,
56}
57
58#[async_trait]
59impl BatchFlusher for TcpFlusher {
60    async fn flush(&self, entries: &[Value]) -> Result<(), FlushError> {
61        let addr = format!("{}:{}", self.host, self.port);
62        let payload = entries_to_lines(entries);
63        let result = async {
64            let mut stream = tokio::time::timeout(self.timeout, TcpStream::connect(&addr))
65                .await
66                .map_err(|_| format!("timed out connecting to TCP server {addr}"))?
67                .map_err(|e| format!("failed to connect to TCP server {addr}: {e}"))?;
68            tokio::time::timeout(self.timeout, stream.write_all(payload.as_bytes()))
69                .await
70                .map_err(|_| format!("timed out sending to TCP server {addr}"))?
71                .map_err(|e| format!("failed to send to TCP server {addr}: {e}"))?;
72            let _ = stream.shutdown().await;
73            Ok::<(), String>(())
74        }
75        .await;
76        result.map_err(|message| FlushError {
77            message,
78            first_fail: None,
79        })
80    }
81}
82
83/// The `error-log-logger` plugin node.
84pub struct ErrorLogLoggerPlugin {
85    sink: BatchSink,
86    log_format: Option<HashMap<String, Value>>,
87}
88
89impl ErrorLogLoggerPlugin {
90    /// Builds the plugin from node config.
91    ///
92    /// Config keys:
93    /// - `host` (string, **required**): TCP sink hostname or IP.
94    /// - `port` (integer, **required**): TCP sink port.
95    /// - `timeout` (integer seconds, default `3`): connect/send timeout.
96    /// - `level` (string, default `"WARN"`): accepted for APISIX compatibility;
97    ///   featherbit logs on the presence of `context.errors`, not on severity.
98    /// - `tls` (bool, default `false`): **not yet supported** — `true` rejected.
99    /// - `log_format` (object): custom `name -> "$var"` entry.
100    /// - batch keys — see [`BatchConfig::from_config`].
101    ///
102    /// ```yaml
103    /// - id: error-log
104    ///   type: error-log-logger
105    ///   config:
106    ///     host: 127.0.0.1
107    ///     port: 5044
108    /// ```
109    pub fn from_config(config: &HashMap<String, Value>) -> Result<Self, String> {
110        let host = config
111            .get("host")
112            .and_then(|v| v.as_str())
113            .filter(|s| !s.is_empty())
114            .ok_or("error-log-logger: `host` is required")?
115            .to_string();
116        let port = config
117            .get("port")
118            .and_then(|v| v.as_u64())
119            .filter(|p| *p <= u16::MAX as u64)
120            .ok_or("error-log-logger: `port` is required and must be 0-65535")?
121            as u16;
122
123        if config.get("tls").and_then(|v| v.as_bool()).unwrap_or(false) {
124            return Err("error-log-logger: TLS (`tls: true`) is not yet supported".to_string());
125        }
126
127        // Validate `level` against the APISIX enum if present (parsed, not enforced).
128        if let Some(level) = config.get("level").and_then(|v| v.as_str()) {
129            const LEVELS: [&str; 10] = [
130                "STDERR", "EMERG", "ALERT", "CRIT", "ERR", "ERROR", "WARN", "NOTICE", "INFO",
131                "DEBUG",
132            ];
133            if !LEVELS.contains(&level) {
134                return Err(format!("error-log-logger: unknown `level` {level}"));
135            }
136        }
137
138        // APISIX error-log-logger `timeout` is in seconds.
139        let timeout_s = config.get("timeout").and_then(|v| v.as_u64()).unwrap_or(3);
140        let timeout = Duration::from_secs(timeout_s.max(1));
141
142        let log_format = parse_log_format(config)?;
143
144        let batch_cfg = BatchConfig::from_config(config)?;
145        let flusher = Arc::new(TcpFlusher {
146            host: host.clone(),
147            port,
148            timeout,
149        });
150        let sink = BatchSink::spawn(
151            &format!("error-log-logger:{host}:{port}"),
152            batch_cfg,
153            flusher,
154        );
155
156        Ok(Self { sink, log_format })
157    }
158}
159
160#[async_trait]
161impl Plugin for ErrorLogLoggerPlugin {
162    fn plugin_type(&self) -> &str {
163        "error-log-logger"
164    }
165
166    async fn execute(&self, ctx: Context, _named_inputs: &HashMap<String, Value>) -> PluginResult {
167        // Only log requests that accumulated errors.
168        if !ctx.errors.is_empty() {
169            let entry = build_entry(&ctx, self.log_format.as_ref(), false, false);
170            self.sink.push(entry);
171        }
172        Ok(PluginOutput {
173            context: ctx,
174            named_outputs: HashMap::new(),
175        })
176    }
177}
178
179#[cfg(test)]
180mod tests {
181    use super::*;
182    use crate::context::{GatewayError, GatewayRequest, GatewayResponse, Protocol};
183    use bytes::Bytes;
184    use serde_json::json;
185
186    fn base_config() -> HashMap<String, Value> {
187        let mut c = HashMap::new();
188        c.insert("host".to_string(), json!("127.0.0.1"));
189        c.insert("port".to_string(), json!(5044));
190        c
191    }
192
193    fn ctx(errors: Vec<GatewayError>) -> Context {
194        Context {
195            request: GatewayRequest {
196                method: "GET".to_string(),
197                path: "/".to_string(),
198                host: "example.com".to_string(),
199                scheme: "http".to_string(),
200                headers: HashMap::new(),
201                query_params: HashMap::new(),
202                body: Bytes::new(),
203                remote_addr: "127.0.0.1:5000".to_string(),
204                protocol: Protocol::Http1,
205            },
206            response: GatewayResponse {
207                status_code: 500,
208                headers: HashMap::new(),
209                body: Bytes::new(),
210            },
211            message: HashMap::new(),
212            errors,
213        }
214    }
215
216    #[test]
217    fn from_config_requires_host_and_port() {
218        assert!(ErrorLogLoggerPlugin::from_config(&HashMap::new()).is_err());
219    }
220
221    #[test]
222    fn from_config_rejects_bad_level_and_tls() {
223        let mut c = base_config();
224        c.insert("level".to_string(), json!("LOUD"));
225        assert!(ErrorLogLoggerPlugin::from_config(&c).is_err());
226
227        let mut c = base_config();
228        c.insert("tls".to_string(), json!(true));
229        assert!(ErrorLogLoggerPlugin::from_config(&c)
230            .err()
231            .unwrap()
232            .contains("TLS"));
233    }
234
235    #[tokio::test]
236    async fn execute_logs_only_on_errors() {
237        // A no-error context produces no entry; an errored one does. We verify
238        // the entry-selection logic via the shared builder rather than the
239        // socket: build_entry over an errored ctx includes an `errors` field.
240        let clean = build_entry(&ctx(vec![]), None, false, false);
241        assert!(clean.get("errors").is_none());
242
243        let errored = ctx(vec![GatewayError {
244            node_id: "upstream".to_string(),
245            code: "502".to_string(),
246            message: "bad gateway".to_string(),
247            metadata: HashMap::new(),
248        }]);
249        let entry = build_entry(&errored, None, false, false);
250        assert!(entry.get("errors").is_some());
251    }
252}
featherbit/plugins/native/error_log_logger.rs

featherbit/plugins/native/
error_log_logger.rs
