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

featherbit/plugins/native/
error_log_logger.rs
