tawasuyu/brahman

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, PathBuf};
14use std::sync::Mutex;
15use tokio::io::unix::AsyncFd;
16use tokio::signal::unix::{signal, SignalKind};
17use tracing::{debug, info, warn};
18use tracing_subscriber::EnvFilter;
19
20const JOURNAL_SOCKET: &str = "/run/systemd/journal/socket";
21const DEV_LOG: &str = "/dev/log";
22
23#[tokio::main(flavor = "current_thread")]
24async fn main() -> anyhow::Result<()> {
25    init_tracing();
26    info!("ente-journald-compat: arrancando");
27    announce_to_fractal().await;
28
29    // Intentamos vincular ambos sockets. Cada uno puede fallar
30    // independientemente; si alguno funciona, seguimos.
31    let mut bound = 0usize;
32    if let Some(stream) = bind_dgram(JOURNAL_SOCKET) {
33        bound += 1;
34        spawn_listener(stream, "journal");
35    } else {
36        warn!(path = JOURNAL_SOCKET, "no se pudo bind — necesita CAP_NET_BIND_SERVICE o /run writable");
37    }
38    if let Some(stream) = bind_dgram(DEV_LOG) {
39        bound += 1;
40        spawn_listener(stream, "syslog");
41    } else {
42        warn!(path = DEV_LOG, "no se pudo bind /dev/log");
43    }
44
45    if bound == 0 {
46        warn!("ningún socket bound — modo idle");
47    } else {
48        info!(sockets_bound = bound, "journald-compat listening");
49    }
50
51    wait_for_term().await
52}
53
54fn bind_dgram(path: &str) -> Option<AsyncFd<OwnedFdWrap>> {
55    use nix::sys::socket::{bind, socket, AddressFamily, SockFlag, SockType, UnixAddr};
56    let _ = std::fs::remove_file(path);
57    if let Some(parent) = Path::new(path).parent() {
58        let _ = std::fs::create_dir_all(parent);
59    }
60    let fd = match socket(
61        AddressFamily::Unix,
62        SockType::Datagram,
63        SockFlag::SOCK_NONBLOCK | SockFlag::SOCK_CLOEXEC,
64        None,
65    ) {
66        Ok(f) => f,
67        Err(e) => { warn!(?e, "socket() falló"); return None; }
68    };
69    let addr = match UnixAddr::new(path) {
70        Ok(a) => a,
71        Err(e) => { warn!(?e, "UnixAddr falló"); return None; }
72    };
73    if let Err(e) = bind(fd.as_raw_fd(), &addr) {
74        warn!(?e, %path, "bind falló");
75        return None;
76    }
77    AsyncFd::new(OwnedFdWrap(fd)).ok()
78}
79
80struct OwnedFdWrap(OwnedFd);
81impl AsRawFd for OwnedFdWrap {
82    fn as_raw_fd(&self) -> std::os::fd::RawFd { self.0.as_raw_fd() }
83}
84
85fn spawn_listener(async_fd: AsyncFd<OwnedFdWrap>, source: &'static str) {
86    tokio::spawn(async move {
87        let mut buf = vec![0u8; 64 * 1024];
88        loop {
89            let mut guard = match async_fd.readable().await {
90                Ok(g) => g,
91                Err(e) => { warn!(?e, source, "readable failed"); return; }
92            };
93            let raw_fd = guard.get_inner().as_raw_fd();
94            loop {
95                let n = unsafe {
96                    libc::recv(raw_fd, buf.as_mut_ptr() as *mut _, buf.len(), 0)
97                };
98                if n <= 0 { break; }
99                handle_message(&buf[..n as usize], source);
100            }
101            guard.clear_ready();
102        }
103    });
104}
105
106/// Mutex sobre el archivo index para escrituras concurrentes desde
107/// múltiples listeners (journal + syslog).
108static INDEX_FILE: Mutex<()> = Mutex::new(());
109
110/// Path del index file: `$XDG_DATA_HOME/ente/journal/index.log` (default
111/// `~/.local/share/ente/journal/index.log`).
112fn index_path() -> PathBuf {
113    let base = if let Ok(d) = std::env::var("XDG_DATA_HOME") { d }
114        else if let Ok(h) = std::env::var("HOME") { format!("{h}/.local/share") }
115        else { "/var/lib".into() };
116    PathBuf::from(base).join("ente").join("journal").join("index.log")
117}
118
119fn now_ms() -> u128 {
120    std::time::SystemTime::now()
121        .duration_since(std::time::UNIX_EPOCH)
122        .map(|d| d.as_millis())
123        .unwrap_or(0)
124}
125
126/// Persiste el blob crudo al CAS y appendea una línea al index:
127/// `<timestamp_ms>:<source>:<unit>:<sha_hex>`. Errores se logean pero
128/// no abortan — perder un mensaje no debe romper journald.
129fn persist_to_cas(buf: &[u8], source: &'static str, unit: Option<&str>) {
130    let sha = match ente_cas::store(buf) {
131        Ok(s) => s,
132        Err(e) => { warn!(?e, "CAS store falló"); return; }
133    };
134    let line = format!(
135        "{}:{}:{}:{}\n",
136        now_ms(), source, unit.unwrap_or("-"), ente_cas::hex(&sha)
137    );
138    let path = index_path();
139    let _guard = INDEX_FILE.lock().unwrap();
140    if let Some(parent) = path.parent() {
141        let _ = std::fs::create_dir_all(parent);
142    }
143    use std::io::Write;
144    let mut f = match std::fs::OpenOptions::new()
145        .create(true).append(true)
146        .open(&path)
147    {
148        Ok(f) => f,
149        Err(e) => { warn!(?e, path = %path.display(), "abrir index"); return; }
150    };
151    if let Err(e) = f.write_all(line.as_bytes()) {
152        warn!(?e, "write index");
153    }
154}
155
156/// Decodifica best-effort. Formato journald nativo: lines de "KEY=value"
157/// (binario para values con newlines, pero raro). Formato syslog: texto
158/// con prefijo "<priority>tag: message".
159fn handle_message(buf: &[u8], source: &'static str) {
160    if let Ok(s) = std::str::from_utf8(buf) {
161        if s.contains('=') && s.lines().any(|l| l.contains('=')) {
162            let mut message = None;
163            let mut priority = None;
164            let mut unit: Option<String> = None;
165            for line in s.lines() {
166                if let Some((k, v)) = line.split_once('=') {
167                    match k {
168                        "MESSAGE" => message = Some(v.to_string()),
169                        "PRIORITY" => priority = Some(v.to_string()),
170                        "_SYSTEMD_UNIT" | "UNIT" => unit = Some(v.to_string()),
171                        _ => {}
172                    }
173                }
174            }
175            persist_to_cas(buf, source, unit.as_deref());
176            if let Some(msg) = message {
177                info!(target: "journal", source, ?priority, ?unit, "{msg}");
178            } else {
179                debug!(source, len = buf.len(), "journal native sin MESSAGE");
180            }
181        } else {
182            persist_to_cas(buf, source, None);
183            info!(target: "syslog", source, "{}", s.trim_end());
184        }
185    } else {
186        persist_to_cas(buf, source, None);
187        debug!(source, len = buf.len(), "journal binario (no UTF-8)");
188    }
189}
190
191async fn announce_to_fractal() {
192    if let Ok(mut client) = BusClient::from_env().await {
193        let req = BusRequest::Announce {
194            capabilities: vec![Capability::Journal],
195        };
196        match client.call(req).await {
197            Ok(BusResponse::Ok) => info!("Announce → bus interno OK"),
198            Ok(other) => warn!(?other, "Announce respuesta inesperada"),
199            Err(e) => warn!(?e, "Announce falló"),
200        }
201    }
202}
203
204async fn wait_for_term() -> anyhow::Result<()> {
205    let mut term = signal(SignalKind::terminate())?;
206    let mut int_ = signal(SignalKind::interrupt())?;
207    tokio::select! {
208        _ = term.recv() => info!("SIGTERM"),
209        _ = int_.recv() => info!("SIGINT"),
210    }
211    Ok(())
212}
213
214fn init_tracing() {
215    let filter = EnvFilter::try_from_default_env()
216        .unwrap_or_else(|_| EnvFilter::new("ente_journald_compat=info,journal=info,syslog=info"));
217    tracing_subscriber::fmt().with_env_filter(filter).with_target(true).init();
218}