rws_server/log/
syslog.rs

1//! Access-log lines to a syslog server (18.33.0, #102): RFC 5424 messages
2//! over UDP, TCP (octet-counting framing, RFC 6587 §3.4.1) or a Unix
3//! datagram socket such as `/dev/log`.
4//!
5//! `RWS_CONFIG_ACCESS_LOG_SYSLOG` = `udp://host:514`, `tcp://host:6514` or
6//! `unix:///dev/log`; `RWS_CONFIG_ACCESS_LOG_SYSLOG_FACILITY` = `local0`
7//! (default) … `local7`, or another facility name or number. Read once, at
8//! the first access-log line. Lines go to syslog **as well as** stdout or
9//! the access-log file, in the configured `RWS_CONFIG_LOG_FORMAT` (JSON is
10//! what SIEMs parse best), severity `info`, APP-NAME `rws`, MSGID `access`.
11//!
12//! Nothing is sent from the request path: the access-log sink hands each
13//! flushed batch (every 100 ms) to a queue that the thread `rws-syslog`
14//! sends. While the server is unreachable batches wait up to
15//! [`MAX_QUEUED`] bytes, then are dropped (counted by [`dropped_lines`]);
16//! a TCP connection is re-made on the next batch.
17
18use std::collections::VecDeque;
19use std::io::Write;
20use std::net::{TcpStream, UdpSocket};
21use std::sync::atomic::{AtomicU64, Ordering};
22use std::sync::{Condvar, Mutex, OnceLock};
23use std::time::Duration;
24
25/// `RWS_CONFIG_ACCESS_LOG_SYSLOG`: where to send access-log lines.
26pub const RWS_CONFIG_ACCESS_LOG_SYSLOG: &str = "RWS_CONFIG_ACCESS_LOG_SYSLOG";
27/// `RWS_CONFIG_ACCESS_LOG_SYSLOG_FACILITY`: the syslog facility (default `local0`).
28pub const RWS_CONFIG_ACCESS_LOG_SYSLOG_FACILITY: &str = "RWS_CONFIG_ACCESS_LOG_SYSLOG_FACILITY";
29
30/// Bytes of unsent lines kept while the server is unreachable.
31pub const MAX_QUEUED: usize = 8 << 20;
32/// Severity of every access-log message: informational.
33const SEVERITY_INFO: u8 = 6;
34
35static DROPPED: AtomicU64 = AtomicU64::new(0);
36
37/// Lines dropped because the queue was full.
38pub fn dropped_lines() -> u64 {
39    DROPPED.load(Ordering::Relaxed)
40}
41
42/// Where messages go.
43#[derive(Clone, Debug, PartialEq, Eq)]
44pub enum Target {
45    /// `udp://host:port`: one datagram per line.
46    Udp(String),
47    /// `tcp://host:port`: octet-counted frames on one connection.
48    Tcp(String),
49    /// `unix:///path`: one datagram per line to a Unix socket.
50    Unix(std::path::PathBuf),
51}
52
53impl Target {
54    /// `udp://host:port`, `tcp://host:port` (port 514 when left out) or
55    /// `unix:///path`.
56    pub fn parse(value: &str) -> Result<Target, String> {
57        let value = value.trim();
58        let with_port = |rest: &str| -> Result<String, String> {
59            let rest = rest.trim_end_matches('/');
60            if rest.is_empty() {
61                return Err(format!("{value}: no host"));
62            }
63            // `[::1]:514`, `host:514`, or a host alone.
64            let has_port = match rest.rfind(':') {
65                Some(i) => rest[i + 1..].parse::<u16>().is_ok() && (!rest.starts_with('[') || rest[..i].ends_with(']')),
66                None => false,
67            };
68            Ok(if has_port { rest.to_string() } else { format!("{rest}:514") })
69        };
70        if let Some(rest) = value.strip_prefix("udp://") {
71            Ok(Target::Udp(with_port(rest)?))
72        } else if let Some(rest) = value.strip_prefix("tcp://") {
73            Ok(Target::Tcp(with_port(rest)?))
74        } else if let Some(path) = value.strip_prefix("unix://") {
75            if !path.starts_with('/') {
76                return Err(format!("{value}: the socket path must be absolute (unix:///dev/log)"));
77            }
78            Ok(Target::Unix(path.into()))
79        } else {
80            Err(format!("{value}: expected udp://host:port, tcp://host:port or unix:///path"))
81        }
82    }
83}
84
85/// A facility name (`kern`, `user`, … `local0`–`local7`) or number 0–23.
86pub fn parse_facility(value: &str) -> Result<u8, String> {
87    const NAMES: [&str; 24] = [
88        "kern", "user", "mail", "daemon", "auth", "syslog", "lpr", "news", "uucp", "cron", "authpriv", "ftp", "ntp", "audit", "alert", "clock",
89        "local0", "local1", "local2", "local3", "local4", "local5", "local6", "local7",
90    ];
91    let value = value.trim().to_ascii_lowercase();
92    if value.is_empty() {
93        return Ok(16);
94    }
95    if let Ok(n) = value.parse::<u8>() {
96        return if n < 24 { Ok(n) } else { Err(format!("syslog facility {n}: must be 0–23")) };
97    }
98    NAMES.iter().position(|n| *n == value).map(|n| n as u8).ok_or_else(|| format!("unknown syslog facility {value}"))
99}
100
101/// Append one RFC 5424 message for `line` to `out`:
102/// `<PRI>1 TIMESTAMP HOSTNAME rws PROCID access - LINE`.
103pub fn message(out: &mut Vec<u8>, pri: u8, timestamp: &str, hostname: &str, pid: u32, line: &[u8]) {
104    let _ = write!(out, "<{pri}>1 {timestamp} {hostname} rws {pid} access - ");
105    out.extend_from_slice(line);
106}
107
108/// This host's name as RFC 5424 allows it (printable ASCII, no spaces,
109/// at most 255 bytes), else `-`.
110fn hostname() -> String {
111    let name = std::env::var("HOSTNAME")
112        .ok()
113        .filter(|h| !h.trim().is_empty())
114        .or_else(|| std::fs::read_to_string("/etc/hostname").ok())
115        .unwrap_or_default();
116    let name: String = name.trim().chars().filter(|c| c.is_ascii_graphic()).take(255).collect();
117    if name.is_empty() { "-".to_string() } else { name }
118}
119
120enum Connection {
121    None,
122    Udp(UdpSocket),
123    Tcp(TcpStream),
124    #[cfg(unix)]
125    Unix(std::os::unix::net::UnixDatagram),
126}
127
128#[derive(Default)]
129struct Queue {
130    batches: VecDeque<Vec<u8>>,
131    bytes: usize,
132}
133
134/// Sends access-log batches to one syslog [`Target`].
135pub struct Sender {
136    target: Target,
137    pri: u8,
138    hostname: String,
139    queue: Mutex<Queue>,
140    wake: Condvar,
141    connection: Mutex<Connection>,
142}
143
144impl Sender {
145    /// A sender for `target` with `facility`; nothing is connected or
146    /// started until something is sent.
147    pub fn new(target: Target, facility: u8) -> Sender {
148        Sender {
149            target,
150            pri: facility * 8 + SEVERITY_INFO,
151            hostname: hostname(),
152            queue: Mutex::new(Queue::default()),
153            wake: Condvar::new(),
154            connection: Mutex::new(Connection::None),
155        }
156    }
157
158    /// Queue a batch of newline-terminated lines (no network I/O here).
159    pub fn enqueue(&self, lines: &[u8]) {
160        if lines.is_empty() {
161            return;
162        }
163        let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
164        if queue.bytes + lines.len() > MAX_QUEUED {
165            DROPPED.fetch_add(memchr::memchr_iter(b'\n', lines).count() as u64, Ordering::Relaxed);
166            return;
167        }
168        queue.bytes += lines.len();
169        queue.batches.push_back(lines.to_vec());
170        drop(queue);
171        self.wake.notify_one();
172    }
173
174    /// Send everything queued now, on this thread.
175    pub fn send_pending(&self) {
176        let batches = {
177            let mut queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
178            queue.bytes = 0;
179            std::mem::take(&mut queue.batches)
180        };
181        if batches.is_empty() {
182            return;
183        }
184        let mut connection = self.connection.lock().unwrap_or_else(|e| e.into_inner());
185        let mut timestamp = String::new();
186        super::iso8601_now(|now| timestamp.push_str(now));
187        let pid = std::process::id();
188        let lines = || batches.iter().flat_map(|b| b.split(|c| *c == b'\n')).filter(|l| !l.is_empty());
189        let result = match self.target {
190            Target::Tcp(_) => {
191                let mut frames = Vec::new();
192                let mut msg = Vec::new();
193                for line in lines() {
194                    msg.clear();
195                    message(&mut msg, self.pri, &timestamp, &self.hostname, pid, line);
196                    let _ = write!(frames, "{} ", msg.len());
197                    frames.extend_from_slice(&msg);
198                }
199                // A connection that died since the last batch fails here;
200                // one more try on a new one.
201                self.send(&mut connection, &frames).or_else(|_| {
202                    *connection = Connection::None;
203                    self.send(&mut connection, &frames)
204                })
205            }
206            _ => {
207                let mut msg = Vec::new();
208                let mut result = Ok(());
209                for line in lines() {
210                    msg.clear();
211                    message(&mut msg, self.pri, &timestamp, &self.hostname, pid, line);
212                    if let Err(e) = self.send(&mut connection, &msg) {
213                        result = Err(e);
214                        *connection = Connection::None;
215                    }
216                }
217                result
218            }
219        };
220        if let Err(e) = result {
221            *connection = Connection::None;
222            complain(&self.target, &e);
223        }
224    }
225
226    /// Send one datagram, or bytes on the stream, connecting first if needed.
227    fn send(&self, connection: &mut Connection, bytes: &[u8]) -> std::io::Result<()> {
228        if matches!(connection, Connection::None) {
229            *connection = self.connect()?;
230        }
231        match connection {
232            Connection::None => Ok(()),
233            Connection::Udp(socket) => socket.send(bytes).map(|_| ()),
234            Connection::Tcp(stream) => stream.write_all(bytes),
235            #[cfg(unix)]
236            Connection::Unix(socket) => match &self.target {
237                Target::Unix(path) => socket.send_to(bytes, path).map(|_| ()),
238                _ => Ok(()),
239            },
240        }
241    }
242
243    fn connect(&self) -> std::io::Result<Connection> {
244        match &self.target {
245            Target::Udp(address) => {
246                let remote = std::net::ToSocketAddrs::to_socket_addrs(address.as_str())?
247                    .next()
248                    .ok_or_else(|| std::io::Error::other(format!("{address}: no address")))?;
249                let local = if remote.is_ipv4() { "0.0.0.0:0" } else { "[::]:0" };
250                let socket = UdpSocket::bind(local)?;
251                socket.connect(remote)?;
252                Ok(Connection::Udp(socket))
253            }
254            Target::Tcp(address) => {
255                let remote = std::net::ToSocketAddrs::to_socket_addrs(address.as_str())?
256                    .next()
257                    .ok_or_else(|| std::io::Error::other(format!("{address}: no address")))?;
258                let stream = TcpStream::connect_timeout(&remote, Duration::from_secs(5))?;
259                stream.set_write_timeout(Some(Duration::from_secs(10)))?;
260                Ok(Connection::Tcp(stream))
261            }
262            #[cfg(unix)]
263            Target::Unix(_) => Ok(Connection::Unix(std::os::unix::net::UnixDatagram::unbound()?)),
264            #[cfg(not(unix))]
265            Target::Unix(_) => Err(std::io::Error::other("unix:// syslog targets need a Unix system")),
266        }
267    }
268
269    /// Send queued batches until the process ends (the `rws-syslog` thread).
270    fn run(&self) {
271        loop {
272            {
273                let queue = self.queue.lock().unwrap_or_else(|e| e.into_inner());
274                let _ = self.wake.wait_timeout_while(queue, Duration::from_secs(1), |q| q.batches.is_empty());
275            }
276            self.send_pending();
277        }
278    }
279}
280
281/// At most one error line a minute.
282fn complain(target: &Target, error: &std::io::Error) {
283    static LAST: Mutex<Option<std::time::Instant>> = Mutex::new(None);
284    let mut last = LAST.lock().unwrap_or_else(|e| e.into_inner());
285    if last.is_none_or(|at| at.elapsed() >= Duration::from_secs(60)) {
286        *last = Some(std::time::Instant::now());
287        eprintln!("access log: syslog {target:?}: {error} (further errors within a minute aren't shown)");
288    }
289}
290
291/// The configured sender, if `RWS_CONFIG_ACCESS_LOG_SYSLOG` is set; its
292/// thread starts with the first call.
293pub fn sender() -> Option<&'static Sender> {
294    static SENDER: OnceLock<Option<Sender>> = OnceLock::new();
295    let sender = SENDER.get_or_init(|| {
296        let value = std::env::var(RWS_CONFIG_ACCESS_LOG_SYSLOG).unwrap_or_default();
297        if value.trim().is_empty() {
298            return None;
299        }
300        let configured = Target::parse(&value).and_then(|target| {
301            let facility = parse_facility(&std::env::var(RWS_CONFIG_ACCESS_LOG_SYSLOG_FACILITY).unwrap_or_default())?;
302            Ok(Sender::new(target, facility))
303        });
304        match configured {
305            Ok(sender) => Some(sender),
306            Err(e) => {
307                eprintln!("access log: {RWS_CONFIG_ACCESS_LOG_SYSLOG}: {e}; not sending to syslog");
308                None
309            }
310        }
311    });
312    let sender = sender.as_ref()?;
313    static STARTED: OnceLock<()> = OnceLock::new();
314    STARTED.get_or_init(|| {
315        let spawned = std::thread::Builder::new().name("rws-syslog".to_string()).spawn(move || sender.run());
316        if spawned.is_err() {
317            eprintln!("access log: no syslog thread; lines are sent at exit only");
318        }
319    });
320    Some(sender)
321}
322
323/// Send what is queued now, on this thread (at process exit).
324pub fn flush_now() {
325    if std::env::var(RWS_CONFIG_ACCESS_LOG_SYSLOG).is_ok_and(|v| !v.trim().is_empty()) {
326        if let Some(sender) = sender() {
327            sender.send_pending();
328        }
329    }
330}
331
332#[cfg(test)]
333mod tests;
rws_server/log/syslog.rs

rws_server/log/
syslog.rs
