1use std::io::Write;
2use std::path::{Path, PathBuf};
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::sync::mpsc::{sync_channel, SyncSender, TrySendError};
5use std::sync::{Arc, OnceLock};
6
7use tracing::Metadata;
8use tracing_subscriber::filter::{filter_fn, FilterExt};
9use tracing_subscriber::layer::SubscriberExt;
10use tracing_subscriber::reload;
11use tracing_subscriber::util::SubscriberInitExt;
12use tracing_subscriber::{EnvFilter, Layer, Registry};
13
14pub static DROPPED: AtomicU64 = AtomicU64::new(0);
15
16type ReloadHandle = reload::Handle<EnvFilter, Registry>;
17static RELOAD: OnceLock<ReloadHandle> = OnceLock::new();
18
19#[derive(Debug, Clone)]
20pub struct LogConfig {
21 pub dir: PathBuf,
22 pub filter: String,
23 pub max_file_bytes: u64,
24 pub max_files: usize,
25 pub queue: usize,
26}
27
28impl Default for LogConfig {
29 fn default() -> Self {
30 LogConfig {
31 dir: PathBuf::from("superfluid-logs"),
32 filter: "info".into(),
33 max_file_bytes: 64 << 20,
34 max_files: 8,
35 queue: 8192,
36 }
37 }
38}
39
40struct Rotating {
41 dir: PathBuf,
42 max_file_bytes: u64,
43 max_files: usize,
44 seq: u64,
45 file: Option<std::fs::File>,
46 written: u64,
47}
48
49impl Rotating {
50 fn open(dir: &Path, max_file_bytes: u64, max_files: usize) -> std::io::Result<Rotating> {
51 std::fs::create_dir_all(dir)?;
52 let mut seq = 0;
53 for e in std::fs::read_dir(dir)?.flatten() {
54 if let Some(n) = e
55 .file_name()
56 .to_str()
57 .and_then(|n| n.strip_prefix("superfluid."))
58 .and_then(|n| n.strip_suffix(".jsonl"))
59 .and_then(|n| n.parse::<u64>().ok())
60 {
61 seq = seq.max(n + 1);
62 }
63 }
64 let mut r = Rotating {
65 dir: dir.to_path_buf(),
66 max_file_bytes,
67 max_files,
68 seq,
69 file: None,
70 written: 0,
71 };
72 r.rotate()?;
73 Ok(r)
74 }
75
76 fn rotate(&mut self) -> std::io::Result<()> {
77 let path = self.dir.join(format!("superfluid.{}.jsonl", self.seq));
78 let mut opts = std::fs::OpenOptions::new();
79 opts.create(true).append(true);
80 #[cfg(unix)]
81 {
82 use std::os::unix::fs::OpenOptionsExt;
83 opts.mode(0o600);
84 }
85 self.file = Some(opts.open(path)?);
86 self.written = 0;
87 self.seq += 1;
88 let mut names: Vec<(u64, PathBuf)> = std::fs::read_dir(&self.dir)?
89 .flatten()
90 .filter_map(|e| {
91 let p = e.path();
92 let n = p
93 .file_name()?
94 .to_str()?
95 .strip_prefix("superfluid.")?
96 .strip_suffix(".jsonl")?
97 .parse::<u64>()
98 .ok()?;
99 Some((n, p))
100 })
101 .collect();
102 names.sort();
103 while names.len() > self.max_files {
104 let (_, p) = names.remove(0);
105 let _ = std::fs::remove_file(p);
106 }
107 Ok(())
108 }
109
110 fn write_line(&mut self, line: &[u8]) {
111 if self.written + line.len() as u64 > self.max_file_bytes && self.rotate().is_err() {
112 DROPPED.fetch_add(1, Ordering::Relaxed);
113 return;
114 }
115 match self.file.as_mut().map(|f| f.write_all(line)) {
116 Some(Ok(())) => self.written += line.len() as u64,
117 _ => {
118 DROPPED.fetch_add(1, Ordering::Relaxed);
119 }
120 }
121 }
122}
123
124#[derive(Clone)]
125struct QueueWriter {
126 tx: SyncSender<Vec<u8>>,
127}
128
129struct LineBuf {
130 tx: SyncSender<Vec<u8>>,
131 buf: Vec<u8>,
132}
133
134impl Write for LineBuf {
135 fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
136 self.buf.extend_from_slice(b);
137 Ok(b.len())
138 }
139 fn flush(&mut self) -> std::io::Result<()> {
140 Ok(())
141 }
142}
143
144impl Drop for LineBuf {
145 fn drop(&mut self) {
146 if self.buf.is_empty() {
147 return;
148 }
149 match self.tx.try_send(std::mem::take(&mut self.buf)) {
150 Ok(()) => {}
151 Err(TrySendError::Full(_)) | Err(TrySendError::Disconnected(_)) => {
152 DROPPED.fetch_add(1, Ordering::Relaxed);
153 }
154 }
155 }
156}
157
158impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for QueueWriter {
159 type Writer = LineBuf;
160 fn make_writer(&'a self) -> Self::Writer {
161 LineBuf {
162 tx: self.tx.clone(),
163 buf: Vec::with_capacity(256),
164 }
165 }
166}
167
168pub type LogRing = std::sync::Arc<std::sync::Mutex<std::collections::VecDeque<String>>>;
169
170const LOG_RING_CAP: usize = 2000;
171
172pub fn new_log_ring() -> LogRing {
173 std::sync::Arc::new(std::sync::Mutex::new(std::collections::VecDeque::with_capacity(256)))
174}
175
176struct RingWriter {
177 ring: LogRing,
178 buf: Vec<u8>,
179}
180impl std::io::Write for RingWriter {
181 fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
182 self.buf.extend_from_slice(b);
183 Ok(b.len())
184 }
185 fn flush(&mut self) -> std::io::Result<()> {
186 Ok(())
187 }
188}
189pub fn push_ring_line(ring: &LogRing, line: String) {
190 if let Ok(mut r) = ring.lock() {
191 while r.len() >= LOG_RING_CAP {
192 r.pop_front();
193 }
194 r.push_back(line);
195 }
196}
197
198impl Drop for RingWriter {
199 fn drop(&mut self) {
200 let line = String::from_utf8_lossy(&self.buf);
201 let line = line.trim_end();
202 if line.is_empty() {
203 return;
204 }
205 if let Ok(mut r) = self.ring.lock() {
206 while r.len() >= LOG_RING_CAP {
207 r.pop_front();
208 }
209 r.push_back(line.to_string());
210 }
211 }
212}
213struct RingMaker {
214 ring: LogRing,
215}
216impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for RingMaker {
217 type Writer = RingWriter;
218 fn make_writer(&'a self) -> RingWriter {
219 RingWriter {
220 ring: self.ring.clone(),
221 buf: Vec::new(),
222 }
223 }
224}
225
226pub fn init(cfg: &LogConfig) -> Result<(), String> {
227 init_sinks(LogSink::File(cfg.clone()), None, None)
228}
229
230pub fn init_stderr(filter: &str) -> Result<(), String> {
231 init_sinks(LogSink::Stderr(filter.to_string()), None, None)
232}
233
234pub fn init_with_ring(cfg: &LogConfig, tui_ring: Option<LogRing>) -> Result<(), String> {
235 init_sinks(LogSink::File(cfg.clone()), tui_ring, None)
236}
237
238#[derive(Debug, Clone)]
239pub enum LogSink {
240 File(LogConfig),
241 Stderr(String),
242}
243
244type BoxedLayer = Box<dyn Layer<Registry> + Send + Sync>;
245
246pub fn init_sinks(
247 sink: LogSink,
248 tui_ring: Option<LogRing>,
249 otlp: Option<crate::otlp::OtlpConfig>,
250) -> Result<(), String> {
251 let mut errors: Vec<String> = Vec::new();
252 let ring_layer = |ring: LogRing| {
253 tracing_subscriber::fmt::layer()
254 .compact()
255 .with_ansi(false)
256 .with_target(false)
257 .with_writer(RingMaker { ring })
258 };
259 let stderr_layer = || {
260 tracing_subscriber::fmt::layer()
261 .compact()
262 .with_target(false)
263 .with_writer(std::io::stderr)
264 };
265 let (logs, filter): (BoxedLayer, String) = match sink {
266 LogSink::File(cfg) => match file_writer(&cfg) {
267 Ok(tx) => {
268 let fmt = tracing_subscriber::fmt::layer()
269 .json()
270 .with_current_span(true)
271 .with_span_list(false)
272 .with_writer(QueueWriter { tx });
273 let layer: BoxedLayer = match tui_ring {
274 Some(ring) => Box::new(fmt.and_then(ring_layer(ring))),
275 None => Box::new(fmt),
276 };
277 (layer, cfg.filter)
278 }
279 Err(e) => {
280 errors.push(format!(
281 "{e} (logging to {})",
282 if tui_ring.is_some() {
283 "the TUI only"
284 } else {
285 "stderr"
286 }
287 ));
288 let layer: BoxedLayer = match tui_ring {
289 Some(ring) => Box::new(ring_layer(ring)),
290 None => Box::new(stderr_layer()),
291 };
292 (layer, cfg.filter)
293 }
294 },
295 LogSink::Stderr(filter) => (Box::new(stderr_layer()), filter),
296 };
297 let filter = EnvFilter::try_new(&filter).unwrap_or_else(|e| {
298 errors.push(format!("log filter {filter:?}: {e} (using \"info\")"));
299 EnvFilter::new("info")
300 });
301 let mut otlp_err = None;
302 let otlp = otlp.and_then(|cfg| {
303 let set_up = EnvFilter::try_new(&cfg.filter)
304 .map_err(|e| format!("--otlp-filter: {e}"))
305 .and_then(|spans| crate::otlp::start(cfg).map(|started| (spans, started)));
306 set_up.map_err(|e| otlp_err = Some(e)).ok()
307 });
308 let (filter, handle) = reload::Layer::new(filter);
309 let reg = tracing_subscriber::registry().with(logs.with_filter(filter));
310 match otlp {
311 Some((spans, (layer, exporter))) => {
312 let only_spans = filter_fn(|m: &Metadata<'_>| m.is_span()).and(spans);
313 reg.with(layer.with_filter(only_spans))
314 .try_init()
315 .map_err(|e| e.to_string())?;
316 crate::otlp::set_global(exporter);
317 }
318 None => reg.try_init().map_err(|e| e.to_string())?,
319 }
320 let _ = RELOAD.set(handle);
321 if let Some(e) = otlp_err {
322 errors.push(format!("OTLP export disabled (local logs unaffected): {e}"));
323 }
324 if errors.is_empty() {
325 Ok(())
326 } else {
327 Err(errors.join("; "))
328 }
329}
330
331fn file_writer(cfg: &LogConfig) -> Result<SyncSender<Vec<u8>>, String> {
332 let mut rot = Rotating::open(&cfg.dir, cfg.max_file_bytes, cfg.max_files)
333 .map_err(|e| format!("log dir {}: {e}", cfg.dir.display()))?;
334 let (tx, rx) = sync_channel::<Vec<u8>>(cfg.queue.max(16));
335 std::thread::Builder::new()
336 .name("superfluid-log-writer".into())
337 .spawn(move || {
338 while let Ok(line) = rx.recv() {
339 rot.write_line(&line);
340 }
341 })
342 .map_err(|e| e.to_string())?;
343 Ok(tx)
344}
345
346pub fn set_filter(directive: &str) -> Result<(), String> {
347 let handle = RELOAD.get().ok_or("telemetry not initialised")?;
348 let filter = EnvFilter::try_new(directive).map_err(|e| e.to_string())?;
349 handle.reload(filter).map_err(|e| e.to_string())
350}
351
352pub fn dropped_total() -> u64 {
353 DROPPED.load(Ordering::Relaxed)
354}
355
356pub type SharedLogConfig = Arc<LogConfig>;