tawasuyu/arje

1//! ente-journald-compat: stub que absorbe escrituras al journal socket.
2//!
3//! Listen en `/run/systemd/journal/socket` (datagram) — todo lo que llega
4//! se decodifica best-effort y se emite como tracing event.
5//!
6//! Sin esto, apps que usan `sd_journal_send` o syslog fallan al escribir.
7//! Para una implementación real: persistir a CAS por timestamp+sha,
8//! exponer query API, indexar por unidad/usuario.
9
10use ente_bus::{BusClient, BusRequest, BusResponse};
11use ente_card::Capability;
12use std::os::fd::{AsRawFd, OwnedFd};
13use std::path::Path;
14use tokio::io::unix::AsyncFd;
15use tokio::signal::unix::{signal, SignalKind};
16use tracing::{debug, info, warn};
17use tracing_subscriber::EnvFilter;
18
19const JOURNAL_SOCKET: &str = "/run/systemd/journal/socket";
20const DEV_LOG: &str = "/dev/log";
21
22#[tokio::main(flavor = "current_thread")]
23async fn main() -> anyhow::Result<()> {
24    init_tracing();
25    info!("ente-journald-compat: arrancando");
26    announce_to_fractal().await;
27
28    // Intentamos vincular ambos sockets. Cada uno puede fallar
29    // independientemente; si alguno funciona, seguimos.
30    let mut bound = 0usize;
31    if let Some(stream) = bind_dgram(JOURNAL_SOCKET) {
32        bound += 1;
33        spawn_listener(stream, "journal");
34    } else {
35        warn!(path = JOURNAL_SOCKET, "no se pudo bind — necesita CAP_NET_BIND_SERVICE o /run writable");
36    }
37    if let Some(stream) = bind_dgram(DEV_LOG) {
38        bound += 1;
39        spawn_listener(stream, "syslog");
40    } else {
41        warn!(path = DEV_LOG, "no se pudo bind /dev/log");
42    }
43
44    if bound == 0 {
45        warn!("ningún socket bound — modo idle");
46    } else {
47        info!(sockets_bound = bound, "journald-compat listening");
48    }
49
50    wait_for_term().await
51}
52
53fn bind_dgram(path: &str) -> Option<AsyncFd<OwnedFdWrap>> {
54    use nix::sys::socket::{bind, socket, AddressFamily, SockFlag, SockType, UnixAddr};
55    let _ = std::fs::remove_file(path);
56    if let Some(parent) = Path::new(path).parent() {
57        let _ = std::fs::create_dir_all(parent);
58    }
59    let fd = match socket(
60        AddressFamily::Unix,
61        SockType::Datagram,
62        SockFlag::SOCK_NONBLOCK | SockFlag::SOCK_CLOEXEC,
63        None,
64    ) {
65        Ok(f) => f,
66        Err(e) => { warn!(?e, "socket() falló"); return None; }
67    };
68    let addr = match UnixAddr::new(path) {
69        Ok(a) => a,
70        Err(e) => { warn!(?e, "UnixAddr falló"); return None; }
71    };
72    if let Err(e) = bind(fd.as_raw_fd(), &addr) {
73        warn!(?e, %path, "bind falló");
74        return None;
75    }
76    AsyncFd::new(OwnedFdWrap(fd)).ok()
77}
78
79struct OwnedFdWrap(OwnedFd);
80impl AsRawFd for OwnedFdWrap {
81    fn as_raw_fd(&self) -> std::os::fd::RawFd { self.0.as_raw_fd() }
82}
83
84fn spawn_listener(async_fd: AsyncFd<OwnedFdWrap>, source: &'static str) {
85    tokio::spawn(async move {
86        let mut buf = vec![0u8; 64 * 1024];
87        loop {
88            let mut guard = match async_fd.readable().await {
89                Ok(g) => g,
90                Err(e) => { warn!(?e, source, "readable failed"); return; }
91            };
92            let raw_fd = guard.get_inner().as_raw_fd();
93            loop {
94                let n = unsafe {
95                    libc::recv(raw_fd, buf.as_mut_ptr() as *mut _, buf.len(), 0)
96                };
97                if n <= 0 { break; }
98                handle_message(&buf[..n as usize], source);
99            }
100            guard.clear_ready();
101        }
102    });
103}
104
105/// Decodifica best-effort. Formato journald nativo: lines de "KEY=value"
106/// (binario para values con newlines, pero raro). Formato syslog: texto
107/// con prefijo "<priority>tag: message".
108fn handle_message(buf: &[u8], source: &'static str) {
109    if let Ok(s) = std::str::from_utf8(buf) {
110        // Heurística: si tiene '=' en alguna línea, asumir journald.
111        if s.contains('=') && s.lines().any(|l| l.contains('=')) {
112            let mut message = None;
113            let mut priority = None;
114            let mut unit = None;
115            for line in s.lines() {
116                if let Some((k, v)) = line.split_once('=') {
117                    match k {
118                        "MESSAGE" => message = Some(v.to_string()),
119                        "PRIORITY" => priority = Some(v.to_string()),
120                        "_SYSTEMD_UNIT" | "UNIT" => unit = Some(v.to_string()),
121                        _ => {}
122                    }
123                }
124            }
125            if let Some(msg) = message {
126                info!(target: "journal", source, ?priority, ?unit, "{msg}");
127            } else {
128                debug!(source, len = buf.len(), "journal native sin MESSAGE");
129            }
130        } else {
131            // Syslog
132            info!(target: "syslog", source, "{}", s.trim_end());
133        }
134    } else {
135        debug!(source, len = buf.len(), "journal binario (no UTF-8)");
136    }
137}
138
139async fn announce_to_fractal() {
140    if let Ok(mut client) = BusClient::from_env().await {
141        let req = BusRequest::Announce {
142            capabilities: vec![Capability::Journal],
143        };
144        match client.call(req).await {
145            Ok(BusResponse::Ok) => info!("Announce → bus interno OK"),
146            Ok(other) => warn!(?other, "Announce respuesta inesperada"),
147            Err(e) => warn!(?e, "Announce falló"),
148        }
149    }
150}
151
152async fn wait_for_term() -> anyhow::Result<()> {
153    let mut term = signal(SignalKind::terminate())?;
154    let mut int_ = signal(SignalKind::interrupt())?;
155    tokio::select! {
156        _ = term.recv() => info!("SIGTERM"),
157        _ = int_.recv() => info!("SIGINT"),
158    }
159    Ok(())
160}
161
162fn init_tracing() {
163    let filter = EnvFilter::try_from_default_env()
164        .unwrap_or_else(|_| EnvFilter::new("ente_journald_compat=info,journal=info,syslog=info"));
165    tracing_subscriber::fmt().with_env_filter(filter).with_target(true).init();
166}