Skip to main content

cortexkit_log/
lib.rs

1//! A synchronous `tracing` layer for CortexKit's module-owned fleet logs.
2//!
3//! One crate, one format, one file per module per day. See
4//! `subconscious/docs/specs/fleet-logging.md` (r2) for the contract every line
5//! here implements; the golden fixture beside it is what the tests pin to.
6
7mod filter;
8mod format;
9mod redaction;
10mod segment;
11mod sink;
12
13pub use segment::{prune_candidates, segment_day, segment_name, SegmentRetention};
14pub use sink::LineSink;
15
16use std::backtrace::Backtrace;
17use std::borrow::Cow;
18use std::env;
19use std::fmt;
20use std::io::{self, Write};
21use std::panic;
22use std::path::{Path, PathBuf};
23use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
24use std::sync::{Arc, Mutex, OnceLock};
25use std::time::SystemTime;
26
27use filter::LevelFilter;
28pub use format::{ParseError, ParsedLevel, ParsedLine};
29use redaction::fleet_redact;
30use segment::{SegmentDestination, SegmentNotice};
31use tracing::field::{Field, Visit};
32use tracing::span::{Attributes, Id, Record};
33use tracing::{Event, Metadata, Subscriber};
34use tracing_subscriber::layer::{Context, SubscriberExt};
35use tracing_subscriber::registry::LookupSpan;
36use tracing_subscriber::{Layer, Registry};
37
38static WRITE_FAILURE_REPORTED: AtomicBool = AtomicBool::new(false);
39static FILTER_FAILURE_REPORTED: AtomicBool = AtomicBool::new(false);
40
41/// Size and age bounds for the rename-rotated [`LineSink`], which has exactly
42/// one writer (the daemon's per-child stderr capture). The logger's own
43/// segments use [`SegmentRetention`]; the two are different files with
44/// different writer counts, and only the single-writer one may rename.
45#[derive(Clone, Copy, Debug, Eq, PartialEq)]
46pub struct Retention {
47    /// Maximum active-file size in mebibytes.
48    pub max_file_mb: u32,
49    /// Number of rotated generations to retain.
50    pub keep: u8,
51    /// Maximum age of rotated generations in days.
52    pub max_age_days: u32,
53}
54
55impl Default for Retention {
56    fn default() -> Self {
57        Self {
58            max_file_mb: 32,
59            keep: 2,
60            max_age_days: 14,
61        }
62    }
63}
64
65impl Retention {
66    const TEST_BYTES_MARKER: u32 = 1 << 31;
67
68    /// Constructs a byte-sized cap so retention tests do not write whole MiB files.
69    #[doc(hidden)]
70    pub fn from_bytes_for_testing(max_file_bytes: u32, keep: u8, max_age_days: u32) -> Self {
71        assert!(max_file_bytes < Self::TEST_BYTES_MARKER);
72        Self {
73            max_file_mb: Self::TEST_BYTES_MARKER | max_file_bytes,
74            keep,
75            max_age_days,
76        }
77    }
78
79    pub(crate) fn max_bytes(self) -> u64 {
80        if self.max_file_mb & Self::TEST_BYTES_MARKER != 0 {
81            return u64::from(self.max_file_mb & !Self::TEST_BYTES_MARKER);
82        }
83
84        u64::from(self.max_file_mb) * 1024 * 1024
85    }
86}
87
88/// A complete-line redactor composed after the fleet credential redactor.
89pub type Redactor = dyn for<'line> Fn(&'line str) -> Cow<'line, str> + Send + Sync;
90
91/// Logger configuration for one process.
92pub struct Config {
93    /// Stable fleet module identifier: the root of every logger name and the
94    /// first part of every segment's file name.
95    pub module_id: String,
96    /// The directory the segments live in. Fleet callers never assemble this:
97    /// [`Config::for_module`] derives it from the module id through the same
98    /// resolver every store uses, so a log cannot land beside the wrong data.
99    /// Public for tests and for the daemon's own `run/logs/` lane, which is not
100    /// a module data directory.
101    pub logs_dir: PathBuf,
102    /// Fields bound for the whole process and rendered in the bracket on every
103    /// line, in this order. A harness-hosted plugin binds `harness=<name>`
104    /// here: after r2 every lane shares one segment and this is what tells
105    /// them apart.
106    pub bound: Vec<(String, String)>,
107    /// `CK_LOG` override; `None` reads the process environment.
108    pub spec: Option<String>,
109    /// Age window and oversize alarm for the segments.
110    pub retention: SegmentRetention,
111    /// Optional module redactor, applied after fleet credential redaction.
112    pub redactor: Option<Arc<Redactor>>,
113    /// Optional clock override for deterministic callers and tests.
114    pub clock: Option<Arc<dyn Fn() -> SystemTime + Send + Sync>>,
115}
116
117impl Config {
118    /// A configuration whose segment directory the caller already resolved:
119    /// `CK_LOG` from the environment, fleet default retention, nothing bound.
120    /// This is the path for a consumer that owns its module data directory
121    /// through its own pinned store crate and must not pull a second copy of
122    /// it through this one (see the `store-paths` feature).
123    pub fn in_dir(module_id: &str, logs_dir: impl Into<PathBuf>) -> Self {
124        Self {
125            module_id: module_id.to_owned(),
126            logs_dir: logs_dir.into(),
127            bound: Vec::new(),
128            spec: None,
129            retention: SegmentRetention::default(),
130            redactor: None,
131            clock: None,
132        }
133    }
134
135    /// The fleet configuration for a supervised module's own process: segments
136    /// under `<module data dir>/logs/`, `CK_LOG` from the environment (the
137    /// daemon injects it at spawn), fleet default retention, nothing bound.
138    #[cfg(feature = "store-paths")]
139    pub fn for_module(module_id: &str) -> Self {
140        let data_dir = PathBuf::from(cortexkit_store_types::module_data_dir(module_id));
141        Self::in_dir(module_id, data_dir.join("logs"))
142    }
143
144    /// [`Config::for_module`] with `harness=<harness>` bound on every line, for
145    /// a plugin running inside a harness process.
146    #[cfg(feature = "store-paths")]
147    pub fn for_plugin(module_id: &str, harness: &str) -> Self {
148        let mut config = Self::for_module(module_id);
149        config
150            .bound
151            .push(("harness".to_owned(), harness.to_owned()));
152        config
153    }
154
155    /// The configuration a supervised module can build with no arguments,
156    /// because the daemon already injects everything it needs at spawn:
157    /// `SUBC_MODULE_ID` (for launch attestation), `CK_LOG`, and the retention
158    /// knobs as `CK_LOG_MAX_AGE_DAYS` / `CK_LOG_ALARM_SEGMENT_MB`. This is the
159    /// zero-argument path the r2 spec requires so that reaching for the crate
160    /// costs no more than reaching for `eprintln!`.
161    #[cfg(feature = "store-paths")]
162    pub fn from_env() -> Result<Self, InitError> {
163        let module_id = env::var("SUBC_MODULE_ID")
164            .ok()
165            .filter(|value| !value.is_empty())
166            .ok_or(InitError::ModuleIdNotInEnvironment)?;
167        let mut config = Self::for_module(&module_id);
168        if let Some(days) = env_u32("CK_LOG_MAX_AGE_DAYS") {
169            config.retention.max_age_days = days;
170        }
171        if let Some(mb) = env_u32("CK_LOG_ALARM_SEGMENT_MB") {
172            config.retention.alarm_segment_mb = mb;
173        }
174        Ok(config)
175    }
176}
177
178#[cfg(feature = "store-paths")]
179fn env_u32(name: &str) -> Option<u32> {
180    env::var(name).ok()?.trim().parse().ok()
181}
182
183/// A live logger handle suitable for inclusion in module health reports.
184#[derive(Clone)]
185pub struct Handle {
186    inner: Arc<LoggerInner>,
187}
188
189impl Handle {
190    /// Returns the number of lines dropped after a write failure.
191    pub fn swallowed_writes(&self) -> u64 {
192        self.inner.swallowed_writes.load(Ordering::Relaxed)
193    }
194
195    /// Returns the segment path for the current instant.
196    pub fn path(&self) -> PathBuf {
197        self.inner.path_now()
198    }
199
200    /// The directory the segments live in.
201    pub fn logs_dir(&self) -> &Path {
202        &self.inner.logs_dir
203    }
204
205    /// Reports whether opening the directory failed and events are going to stderr.
206    pub fn fallback_active(&self) -> bool {
207        self.inner.fallback_active
208    }
209
210    /// Writes one line stamped with `at` into the segment `at` names, instead
211    /// of the one the logger's clock names now.
212    ///
213    /// For a record that must sit beside lines written under a different clock
214    /// reading. The case it exists for is a wall-clock step: the lines before
215    /// the step went to the segment the old clock named, so a marker describing
216    /// them belongs there too, stamped the way its neighbours are, or a reader
217    /// opening that file finds the lines and not the warning. Not level-filtered
218    /// (CK_LOG does not apply) and not for routine logging, which goes through
219    /// `tracing` so filtering and span context hold.
220    pub fn emit_at(
221        &self,
222        at: SystemTime,
223        level: tracing::Level,
224        logger: &str,
225        message: &str,
226        fields: &[(String, String)],
227    ) {
228        self.inner.emit_at(at, &level, logger, &[], message, fields);
229    }
230
231    /// [`Handle::emit_at`] with bound fields too, for a writer that renders on a
232    /// thread other than the one that produced the event (for example a queue
233    /// drained by a dedicated log thread). Such a writer cannot rely on the
234    /// caller's `tracing` span, so it carries the context itself: `bound` is
235    /// rendered in the bracket after the process-level fields, replacing a
236    /// process-level field of the same key. Not level-filtered (`CK_LOG` does
237    /// not apply); the producer decides what to enqueue.
238    pub fn emit_at_with_bound(
239        &self,
240        at: SystemTime,
241        level: tracing::Level,
242        logger: &str,
243        bound: &[(String, String)],
244        message: &str,
245        fields: &[(String, String)],
246    ) {
247        self.inner
248            .emit_at(at, &level, logger, bound, message, fields);
249    }
250
251    /// Installs the panic hook [`init`] installs: the panic text and a forced
252    /// backtrace go into the segment, one line each on logger
253    /// `<module>.panic`, and the previous hook still runs (the default one
254    /// prints to stderr). For a caller that composed the layer itself with
255    /// [`layer`] or [`layer_with_stderr_copy`]. Call it once; each call chains
256    /// another hook.
257    pub fn install_panic_hook(&self) {
258        install_panic_hook(Arc::clone(&self.inner));
259    }
260}
261
262// The handle of the logger `init` installed, so code that did not install it
263// (a library running inside a module's process) can still reach
264// `Handle::emit_at`. `init` can succeed only once per process, since the
265// tracing global can only be set once, so there is at most one to hold.
266static INSTALLED: OnceLock<Handle> = OnceLock::new();
267
268/// The logger this process installed with [`init`], if it installed one.
269pub fn installed() -> Option<Handle> {
270    INSTALLED.get().cloned()
271}
272
273/// An error that prevents installation of the global logger.
274#[derive(Clone, Debug, Eq, PartialEq)]
275pub enum InitError {
276    /// [`Config::from_env`] found no `SUBC_MODULE_ID`; the process is not a
277    /// daemon-spawned module and must name itself.
278    ModuleIdNotInEnvironment,
279    /// A process-global tracing subscriber was already installed.
280    GlobalSubscriberAlreadySet,
281}
282
283impl fmt::Display for InitError {
284    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
285        match self {
286            Self::ModuleIdNotInEnvironment => formatter.write_str(
287                "SUBC_MODULE_ID is not set; use Config::for_module(<id>) outside supervision",
288            ),
289            Self::GlobalSubscriberAlreadySet => {
290                formatter.write_str("a global tracing subscriber is already installed")
291            }
292        }
293    }
294}
295
296impl std::error::Error for InitError {}
297
298/// Installs the fleet logger as the process-global `tracing` subscriber.
299pub fn init(config: Config) -> Result<Handle, InitError> {
300    let (layer, handle) = build_layer(config, Box::new(io::stderr()), false)?;
301    let panic_inner = Arc::clone(&handle.inner);
302    let subscriber = Registry::default().with(layer);
303    tracing::subscriber::set_global_default(subscriber)
304        .map_err(|_| InitError::GlobalSubscriberAlreadySet)?;
305    install_panic_hook(panic_inner);
306    let _ = INSTALLED.set(handle.clone());
307    Ok(handle)
308}
309
310/// The fleet layer without installing anything global, for a caller that
311/// composes its own `tracing` subscriber stack. Unlike [`init`], it installs no
312/// global subscriber and no panic hook ([`Handle::install_panic_hook`] does
313/// that), and [`installed`] does not return its handle.
314pub fn layer(config: Config) -> Result<(LogLayer, Handle), InitError> {
315    build_layer(config, Box::new(io::stderr()), false)
316}
317
318/// [`layer`] that also writes every rendered line to stderr after it reaches
319/// the segment. For a process whose stderr is watched by a person or a parent
320/// (a standalone bridge, a CLI), where the segment alone would hide its output.
321/// A supervised module should not use it: the daemon captures a module's stderr
322/// separately, so every line would be stored twice. When the segment directory
323/// cannot be opened, lines already go to stderr and are not doubled.
324pub fn layer_with_stderr_copy(config: Config) -> Result<(LogLayer, Handle), InitError> {
325    build_layer(config, Box::new(io::stderr()), true)
326}
327
328/// [`init`] with [`Config::from_env`]: the whole setup for a supervised module.
329#[cfg(feature = "store-paths")]
330pub fn init_from_env() -> Result<Handle, InitError> {
331    init(Config::from_env()?)
332}
333
334/// Creates a span carrying canonical session lineage for nested events. Every
335/// event inside it renders `session=<issuer>:<id>` in the bound bracket.
336pub fn session_span(issuer: &str, id: &str) -> tracing::Span {
337    // An empty half means "no session": the line carries no field rather
338    // than a placeholder. Callers that used to log a synthetic id must pass
339    // nothing here instead; the crate does not know their sentinels.
340    if issuer.is_empty() || id.is_empty() {
341        tracing::Span::none()
342    } else {
343        let session = format!("{issuer}:{id}");
344        tracing::info_span!("cortexkit.session", session = %session)
345    }
346}
347
348/// Parses the fixed columns used by merged and filtered fleet log views.
349pub fn parse_line(line: &str) -> Result<ParsedLine<'_>, ParseError> {
350    format::parse(line)
351}
352
353struct LoggerInner {
354    module_id: String,
355    logs_dir: PathBuf,
356    process_bound: Vec<(String, String)>,
357    destination: Mutex<Option<SegmentDestination>>,
358    stderr: Mutex<Box<dyn Write + Send>>,
359    swallowed_writes: AtomicU64,
360    fallback_active: bool,
361    stderr_copy: bool,
362    redactor: Option<Arc<Redactor>>,
363    clock: Arc<dyn Fn() -> SystemTime + Send + Sync>,
364}
365
366impl LoggerInner {
367    fn path_now(&self) -> PathBuf {
368        self.logs_dir
369            .join(segment::segment_name(&self.module_id, (self.clock)()))
370    }
371
372    fn emit(
373        &self,
374        level: &tracing::Level,
375        logger: &str,
376        scoped_bound: &[(String, String)],
377        message: &str,
378        fields: &[(String, String)],
379    ) {
380        self.emit_at((self.clock)(), level, logger, scoped_bound, message, fields);
381    }
382
383    fn emit_at(
384        &self,
385        at: SystemTime,
386        level: &tracing::Level,
387        logger: &str,
388        scoped_bound: &[(String, String)],
389        message: &str,
390        fields: &[(String, String)],
391    ) {
392        // Process-level context first, then scoped: the process-level part is
393        // identical on every line the process writes, so putting it first keeps
394        // the bracket's leading columns stable within a file.
395        let mut bound: Vec<(String, String)> = self.process_bound.clone();
396        for (key, value) in scoped_bound {
397            match bound.iter_mut().find(|(existing, _)| existing == key) {
398                Some(slot) => slot.1 = value.clone(),
399                None => bound.push((key.clone(), value.clone())),
400            }
401        }
402        let raw = format::render_line(at, level, logger, &bound, message, fields);
403        let fleet_redacted = fleet_redact(&raw);
404        let module_redacted = self.redactor.as_ref().map_or_else(
405            || Cow::Borrowed(fleet_redacted.as_ref()),
406            |redactor| redactor(&fleet_redacted),
407        );
408        // Redactors are extensibility points, so the final guard preserves the
409        // one-line, no-raw-control contract even if a module redactor introduces
410        // such bytes. It escapes rather than strips, because a strip can consume
411        // text the redactor never meant to touch.
412        let guarded = format::escape_raw_controls(&module_redacted);
413        self.write_line(&guarded, at);
414    }
415
416    fn write_line(&self, line: &str, at: SystemTime) {
417        let mut bytes = Vec::with_capacity(line.len() + 1);
418        bytes.extend_from_slice(line.as_bytes());
419        bytes.push(b'\n');
420
421        // The mutex keeps this process's lines whole against its own threads;
422        // O_APPEND keeps them whole against other processes. No queue, no
423        // background flusher: a caller waits only for the write in front of it.
424        let result = {
425            let mut destination = self
426                .destination
427                .lock()
428                .unwrap_or_else(std::sync::PoisonError::into_inner);
429            match destination.as_mut() {
430                None => {
431                    let mut stderr = self
432                        .stderr
433                        .lock()
434                        .unwrap_or_else(std::sync::PoisonError::into_inner);
435                    stderr.write_all(&bytes).map(|()| None)
436                }
437                Some(segment) => {
438                    let written = segment.write(&bytes, at);
439                    if self.stderr_copy && written.is_ok() {
440                        let mut stderr = self
441                            .stderr
442                            .lock()
443                            .unwrap_or_else(std::sync::PoisonError::into_inner);
444                        let _ = stderr.write_all(&bytes);
445                    }
446                    written
447                }
448            }
449        };
450
451        match result {
452            Ok(None) => {}
453            Ok(Some(notice)) => self.report_notice(notice),
454            Err(error) => {
455                self.swallowed_writes.fetch_add(1, Ordering::Relaxed);
456                if !WRITE_FAILURE_REPORTED.swap(true, Ordering::Relaxed) {
457                    self.report(&format!(
458                        "cortexkit-log: log write failed; future failures will be swallowed: {error}\n"
459                    ));
460                }
461            }
462        }
463    }
464
465    // A feature that fires later must say that it fired: without these lines,
466    // "retention never ran" and "ran and found nothing to prune" are the same
467    // observation from outside. Oversize is reported once per process because
468    // a runaway writer must not get a second flood from its own alarm.
469    fn report_notice(&self, notice: SegmentNotice) {
470        match notice {
471            SegmentNotice::Pruned { removed, kept } => self.report(&format!(
472                "cortexkit-log: {}.retention pruned={removed} kept={kept}\n",
473                self.module_id
474            )),
475            SegmentNotice::Oversized { path, bytes } => self.report(&format!(
476                "cortexkit-log: segment oversized, NOT truncated: path={} bytes={bytes}\n",
477                path.display()
478            )),
479        }
480    }
481
482    fn report(&self, report: &str) {
483        let mut stderr = self
484            .stderr
485            .lock()
486            .unwrap_or_else(std::sync::PoisonError::into_inner);
487        let _ = stderr.write_all(report.as_bytes());
488    }
489
490    fn write_panic(&self, information: &panic::PanicHookInfo<'_>) {
491        let at = (self.clock)();
492        let logger = format!("{}.panic", self.module_id);
493        let panic_text = information.to_string();
494        for line in panic_text.lines() {
495            self.emit_at(at, &tracing::Level::ERROR, &logger, &[], line, &[]);
496        }
497        let backtrace = Backtrace::force_capture().to_string();
498        for line in backtrace.lines() {
499            self.emit_at(at, &tracing::Level::ERROR, &logger, &[], line, &[]);
500        }
501    }
502}
503
504/// The fleet `tracing` layer, built by [`layer`] or [`layer_with_stderr_copy`]
505/// for a caller that installs its own subscriber.
506pub struct LogLayer {
507    inner: Arc<LoggerInner>,
508    filter: LevelFilter,
509}
510
511impl<S> Layer<S> for LogLayer
512where
513    S: Subscriber + for<'lookup> LookupSpan<'lookup>,
514{
515    fn enabled(&self, metadata: &Metadata<'_>, _context: Context<'_, S>) -> bool {
516        if metadata.is_span() {
517            return true;
518        }
519        let logger = format::logger_name(&self.inner.module_id, metadata.target());
520        self.filter.enabled(&logger, metadata.level())
521    }
522
523    fn on_new_span(&self, attributes: &Attributes<'_>, id: &Id, context: Context<'_, S>) {
524        let mut visitor = BoundVisitor::default();
525        attributes.record(&mut visitor);
526        if let (false, Some(span)) = (visitor.bound.is_empty(), context.span(id)) {
527            span.extensions_mut().insert(SpanBound(visitor.bound));
528        }
529    }
530
531    fn on_record(&self, id: &Id, values: &Record<'_>, context: Context<'_, S>) {
532        let mut visitor = BoundVisitor::default();
533        values.record(&mut visitor);
534        if let (false, Some(span)) = (visitor.bound.is_empty(), context.span(id)) {
535            let mut extensions = span.extensions_mut();
536            match extensions.get_mut::<SpanBound>() {
537                Some(existing) => existing.0.extend(visitor.bound),
538                None => extensions.insert(SpanBound(visitor.bound)),
539            }
540        }
541    }
542
543    fn on_event(&self, event: &Event<'_>, context: Context<'_, S>) {
544        let mut visitor = EventVisitor::default();
545        event.record(&mut visitor);
546        // Every field on every span enclosing the event is bound context,
547        // root-first so an inner span overrides an outer one with the same key.
548        // This is MDC: the module decides what is in scope by what it puts on
549        // its spans, and the crate renders whatever is there.
550        let mut scoped: Vec<(String, String)> = Vec::new();
551        if let Some(scope) = context.event_scope(event) {
552            for span in scope.from_root() {
553                if let Some(bound) = span.extensions().get::<SpanBound>() {
554                    for (key, value) in &bound.0 {
555                        match scoped.iter_mut().find(|(existing, _)| existing == key) {
556                            Some(slot) => slot.1 = value.clone(),
557                            None => scoped.push((key.clone(), value.clone())),
558                        }
559                    }
560                }
561            }
562        }
563        let logger = format::logger_name(&self.inner.module_id, event.metadata().target());
564        self.inner.emit(
565            event.metadata().level(),
566            &logger,
567            &scoped,
568            visitor.message.as_deref().unwrap_or(""),
569            &visitor.fields,
570        );
571    }
572}
573
574#[derive(Clone)]
575struct SpanBound(Vec<(String, String)>);
576
577#[derive(Default)]
578struct BoundVisitor {
579    bound: Vec<(String, String)>,
580}
581
582impl BoundVisitor {
583    fn record(&mut self, field: &Field, value: String) {
584        // An empty value binds nothing: `session_span("", "")` and a plugin
585        // with no harness must produce a line with no bracket, never `key=`.
586        if !value.is_empty() {
587            self.bound.push((field.name().to_owned(), value));
588        }
589    }
590}
591
592impl Visit for BoundVisitor {
593    fn record_str(&mut self, field: &Field, value: &str) {
594        self.record(field, value.to_owned());
595    }
596
597    fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) {
598        let rendered = format!("{value:?}");
599        let value = rendered
600            .strip_prefix('"')
601            .and_then(|unquoted| unquoted.strip_suffix('"'))
602            .unwrap_or(&rendered);
603        self.record(field, value.to_owned());
604    }
605
606    fn record_i64(&mut self, field: &Field, value: i64) {
607        self.record(field, value.to_string());
608    }
609
610    fn record_u64(&mut self, field: &Field, value: u64) {
611        self.record(field, value.to_string());
612    }
613
614    fn record_bool(&mut self, field: &Field, value: bool) {
615        self.record(field, value.to_string());
616    }
617}
618
619#[derive(Default)]
620struct EventVisitor {
621    message: Option<String>,
622    fields: Vec<(String, String)>,
623}
624
625impl EventVisitor {
626    fn record(&mut self, field: &Field, value: String) {
627        if field.name() == "message" {
628            self.message = Some(value);
629        } else {
630            self.fields.push((field.name().to_owned(), value));
631        }
632    }
633}
634
635impl Visit for EventVisitor {
636    fn record_f64(&mut self, field: &Field, value: f64) {
637        self.record(field, value.to_string());
638    }
639
640    fn record_i64(&mut self, field: &Field, value: i64) {
641        self.record(field, value.to_string());
642    }
643
644    fn record_u64(&mut self, field: &Field, value: u64) {
645        self.record(field, value.to_string());
646    }
647
648    fn record_i128(&mut self, field: &Field, value: i128) {
649        self.record(field, value.to_string());
650    }
651
652    fn record_u128(&mut self, field: &Field, value: u128) {
653        self.record(field, value.to_string());
654    }
655
656    fn record_bool(&mut self, field: &Field, value: bool) {
657        self.record(field, value.to_string());
658    }
659
660    fn record_str(&mut self, field: &Field, value: &str) {
661        self.record(field, value.to_owned());
662    }
663
664    fn record_error(&mut self, field: &Field, value: &(dyn std::error::Error + 'static)) {
665        self.record(field, value.to_string());
666    }
667
668    fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) {
669        self.record(field, format!("{value:?}"));
670    }
671}
672
673fn build_layer(
674    config: Config,
675    stderr: Box<dyn Write + Send>,
676    stderr_copy: bool,
677) -> Result<(LogLayer, Handle), InitError> {
678    let clock = config.clock.unwrap_or_else(|| Arc::new(SystemTime::now));
679    let now = clock();
680    let (destination, open_notice, open_error) = match SegmentDestination::open(
681        &config.logs_dir,
682        &config.module_id,
683        config.retention,
684        now,
685        true,
686    ) {
687        Ok((destination, notice)) => (Some(destination), notice, None),
688        Err(error) => (None, None, Some(error)),
689    };
690    let fallback_active = open_error.is_some();
691    let inner = Arc::new(LoggerInner {
692        module_id: config.module_id,
693        logs_dir: config.logs_dir,
694        process_bound: config.bound,
695        destination: Mutex::new(destination),
696        stderr: Mutex::new(stderr),
697        swallowed_writes: AtomicU64::new(0),
698        fallback_active,
699        stderr_copy,
700        redactor: config.redactor,
701        clock,
702    });
703
704    if let Some(notice) = open_notice {
705        inner.report_notice(notice);
706    }
707    if let Some(error) = open_error {
708        let logger = inner.module_id.clone();
709        inner.emit(
710            &tracing::Level::ERROR,
711            &logger,
712            &[],
713            "log directory unavailable; falling back to stderr",
714            &[
715                ("dir".to_owned(), inner.logs_dir.display().to_string()),
716                ("error".to_owned(), error.to_string()),
717            ],
718        );
719    }
720
721    let filter = make_filter(config.spec, &inner);
722    let layer = LogLayer {
723        inner: Arc::clone(&inner),
724        filter,
725    };
726    Ok((layer, Handle { inner }))
727}
728
729fn make_filter(spec_override: Option<String>, inner: &LoggerInner) -> LevelFilter {
730    let spec = spec_override.or_else(|| env::var("CK_LOG").ok());
731    let spec = spec
732        .as_deref()
733        .map(str::trim)
734        .filter(|value| !value.is_empty());
735    match spec {
736        Some(spec) => LevelFilter::parse(spec).unwrap_or_else(|error| {
737            if !FILTER_FAILURE_REPORTED.swap(true, Ordering::Relaxed) {
738                inner.report(&format!(
739                    "cortexkit-log: invalid CK_LOG value {spec:?}; using info: {error}\n"
740                ));
741            }
742            LevelFilter::info()
743        }),
744        None => LevelFilter::info(),
745    }
746}
747
748fn install_panic_hook(inner: Arc<LoggerInner>) {
749    let previous = panic::take_hook();
750    panic::set_hook(Box::new(move |information| {
751        inner.write_panic(information);
752        previous(information);
753    }));
754}
755
756#[cfg(test)]
757mod tests;