1mod 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#[derive(Clone, Copy, Debug, Eq, PartialEq)]
46pub struct Retention {
47 pub max_file_mb: u32,
49 pub keep: u8,
51 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 #[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
88pub type Redactor = dyn for<'line> Fn(&'line str) -> Cow<'line, str> + Send + Sync;
90
91pub struct Config {
93 pub module_id: String,
96 pub logs_dir: PathBuf,
102 pub bound: Vec<(String, String)>,
107 pub spec: Option<String>,
109 pub retention: SegmentRetention,
111 pub redactor: Option<Arc<Redactor>>,
113 pub clock: Option<Arc<dyn Fn() -> SystemTime + Send + Sync>>,
115}
116
117impl Config {
118 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 #[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 #[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 #[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#[derive(Clone)]
185pub struct Handle {
186 inner: Arc<LoggerInner>,
187}
188
189impl Handle {
190 pub fn swallowed_writes(&self) -> u64 {
192 self.inner.swallowed_writes.load(Ordering::Relaxed)
193 }
194
195 pub fn path(&self) -> PathBuf {
197 self.inner.path_now()
198 }
199
200 pub fn logs_dir(&self) -> &Path {
202 &self.inner.logs_dir
203 }
204
205 pub fn fallback_active(&self) -> bool {
207 self.inner.fallback_active
208 }
209
210 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 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 pub fn install_panic_hook(&self) {
258 install_panic_hook(Arc::clone(&self.inner));
259 }
260}
261
262static INSTALLED: OnceLock<Handle> = OnceLock::new();
267
268pub fn installed() -> Option<Handle> {
270 INSTALLED.get().cloned()
271}
272
273#[derive(Clone, Debug, Eq, PartialEq)]
275pub enum InitError {
276 ModuleIdNotInEnvironment,
279 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
298pub 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
310pub fn layer(config: Config) -> Result<(LogLayer, Handle), InitError> {
315 build_layer(config, Box::new(io::stderr()), false)
316}
317
318pub fn layer_with_stderr_copy(config: Config) -> Result<(LogLayer, Handle), InitError> {
325 build_layer(config, Box::new(io::stderr()), true)
326}
327
328#[cfg(feature = "store-paths")]
330pub fn init_from_env() -> Result<Handle, InitError> {
331 init(Config::from_env()?)
332}
333
334pub fn session_span(issuer: &str, id: &str) -> tracing::Span {
337 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
348pub 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 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 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 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 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
504pub 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 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 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;