dig_logging/
init.rs

1//! [`init`] — the one entry point that installs the DIG logging stack (SPEC §1).
2//!
3//! It resolves the log directory, opens the rolling file appender, wires the JSONL file sink + the
4//! human stderr sink behind one reloadable level filter, stamps the correlation ids, and spawns the
5//! hourly maintenance task (byte-cap janitor + dropped-line reporter). It returns a [`LogGuard`] the
6//! caller holds for the process lifetime; dropping it flushes the file writer.
7
8use std::path::PathBuf;
9use std::time::Duration;
10
11use tracing_appender::rolling::{Builder as AppenderBuilder, Rotation};
12use tracing_subscriber::layer::SubscriberExt;
13use tracing_subscriber::util::SubscriberInitExt;
14use tracing_subscriber::{reload, EnvFilter, Layer};
15
16use crate::error::{Error, Result};
17use crate::layer::{DigJsonLayer, OwnedStatics};
18use crate::writer::{LossyWriter, WriterGuard};
19use crate::{correlation, dirs, filter, janitor, Service};
20
21/// How often the maintenance task enforces the byte cap + reports dropped lines.
22const MAINTENANCE_INTERVAL: Duration = Duration::from_secs(3600);
23
24/// A type-erased live-filter swapper capturing the `reload` handle (SPEC §5 runtime reload).
25type FilterSetter = Box<dyn Fn(&str) -> Result<()> + Send + Sync>;
26
27/// Held by the caller for the life of the process. Dropping it flushes the file writer (SPEC §4.4);
28/// it also exposes runtime level control (SPEC §5).
29pub struct LogGuard {
30    _writer: WriterGuard,
31    dir: PathBuf,
32    set_filter: FilterSetter,
33}
34
35impl LogGuard {
36    /// The resolved log directory this run is writing to.
37    pub fn log_dir(&self) -> &std::path::Path {
38        &self.dir
39    }
40
41    /// Swap the live level filter (SPEC §5 runtime reload). The consumer's control plane calls this.
42    pub fn set_filter(&self, directive: &str) -> Result<()> {
43        (self.set_filter)(directive)
44    }
45}
46
47/// Install the DIG logging stack for `service` (SPEC §1). Call ONCE at process start; a second call
48/// returns [`Error::AlreadyInitialized`].
49pub fn init(service: Service) -> Result<LogGuard> {
50    let dir = dirs::log_dir(service.name);
51    std::fs::create_dir_all(&dir).map_err(|source| Error::LogDir {
52        path: dir.clone(),
53        source,
54    })?;
55
56    let env = |key: &str| std::env::var(key).ok();
57    let retention = janitor::retention_days(env);
58    let max_bytes = janitor::max_bytes(env);
59    janitor::enforce_byte_cap(&dir, service.name, max_bytes);
60
61    let appender = AppenderBuilder::new()
62        .rotation(Rotation::DAILY)
63        .filename_prefix(format!("{}.jsonl", service.name))
64        .max_log_files(retention)
65        .build(&dir)
66        .map_err(|source| Error::Appender {
67            path: dir.clone(),
68            source,
69        })?;
70    let (file_writer, writer_guard) = crate::writer::spawn(appender);
71
72    let statics = OwnedStatics {
73        service: service.name.to_string(),
74        service_version: service.version.to_string(),
75        run_context: service.run_context.as_str().to_string(),
76        run_id: correlation::new_run_id(),
77        parent_op_id: correlation::parent_op_id_from_env(),
78    };
79
80    let directive = filter::resolve_filter_from_env(filter::read_persisted_level(&dir).as_deref());
81    let env_filter = EnvFilter::try_new(&directive).map_err(|e| Error::Filter {
82        directive: directive.clone(),
83        message: e.to_string(),
84    })?;
85    let (filter_layer, reload_handle) = reload::Layer::new(env_filter);
86
87    let json_layer = DigJsonLayer::new(statics, file_writer.clone());
88    let stderr_layer = tracing_subscriber::fmt::layer()
89        .with_writer(std::io::stderr)
90        .compact();
91
92    tracing_subscriber::registry()
93        .with(filter_layer)
94        .with(json_layer)
95        .with(stderr_layer.boxed())
96        .try_init()
97        .map_err(|_| Error::AlreadyInitialized)?;
98
99    spawn_maintenance(dir.clone(), service.name, max_bytes, file_writer);
100
101    let set_filter = Box::new(move |directive: &str| -> Result<()> {
102        let new = EnvFilter::try_new(directive).map_err(|e| Error::Filter {
103            directive: directive.to_string(),
104            message: e.to_string(),
105        })?;
106        reload_handle.reload(new).map_err(|e| Error::Filter {
107            directive: directive.to_string(),
108            message: e.to_string(),
109        })
110    });
111
112    Ok(LogGuard {
113        _writer: writer_guard,
114        dir,
115        set_filter,
116    })
117}
118
119/// Spawn the hourly maintenance task: enforce the byte cap, and emit a `WARN` whenever the file
120/// writer has dropped new lines under backpressure (SPEC §4/§4.4).
121fn spawn_maintenance(dir: PathBuf, service: &'static str, max_bytes: u64, writer: LossyWriter) {
122    std::thread::Builder::new()
123        .name("dig-logging-maintenance".into())
124        .spawn(move || {
125            let mut last_dropped = 0u64;
126            loop {
127                std::thread::sleep(MAINTENANCE_INTERVAL);
128                janitor::enforce_byte_cap(&dir, service, max_bytes);
129                let dropped = writer.dropped();
130                if dropped > last_dropped {
131                    tracing::warn!(
132                        target: "dig_logging",
133                        dropped,
134                        "log lines dropped under backpressure since start"
135                    );
136                    last_dropped = dropped;
137                }
138            }
139        })
140        .expect("spawn dig-logging maintenance thread");
141}
dig_logging/init.rs

dig_logging/
init.rs
