1use std::{collections::HashMap, path::PathBuf, time::Duration};
19
20use chrono::SecondsFormat;
21use eyre::{DefaultHandler, EyreHandler, bail};
22use tracing::Subscriber;
23use tracing_appender::rolling::RollingFileAppender;
24use tracing_subscriber::{
25 EnvFilter, Layer, Registry,
26 layer::SubscriberExt,
27 registry,
28 reload::{self, Handle},
29};
30
31#[cfg(feature = "log_throttling")]
32use tracing_throttle::TracingRateLimitLayer;
33
34#[cfg(feature = "error_aggregation")]
35use {crate::libs::log::error_aggregation::*, std::sync::Arc};
36
37#[deprecated(
38 since = "1.3.0",
39 note = "This code is not used in current projects and should probably not be used going forward"
40)]
41pub mod legacy;
42
43#[cfg(feature = "error_aggregation")]
44pub mod error_aggregation;
45pub mod level_filter;
46
47pub use level_filter::*;
48
49pub use tracing_appender::rolling::Rotation as LogRotation;
51
52pub use tracing_appender::non_blocking::WorkerGuard;
53
54#[derive(Debug, Clone, Default)]
68pub struct OtelConfig {
69 pub enabled: bool,
71 pub service_name: Option<String>,
73 pub endpoint: Option<String>,
75 pub headers: HashMap<String, String>,
77}
78
79#[derive(Debug)]
80pub struct LoggingConfig {
81 pub level: LogLevel,
82 pub file_config: Option<FileLoggingConfig>,
83 pub otel_config: OtelConfig,
87 #[cfg(feature = "error_aggregation")]
88 pub error_aggregation: ErrorAggregationConfig,
89 #[cfg(feature = "log_throttling")]
90 pub throttling_config: Option<LogThrottlingConfig>,
91}
92
93#[derive(Debug, Clone)]
94pub struct FileLoggingConfig {
95 pub path: PathBuf,
96 pub file_prefix: Option<String>,
99 pub rotation: Option<LogRotation>,
101}
102
103#[derive(Debug, Default, Clone)]
104pub struct LogThrottlingConfig {
105 pub summary_emission_interval: Option<Duration>,
107 pub metrics_emission_interval: Option<Duration>,
109 pub excluded_fields: Option<Vec<String>>,
117 pub exemptions: Option<Vec<String>>,
126}
127
128pub struct LogSetupReturn {
129 pub reload_handle: LogReloadHandle,
130 pub log_guards: (WorkerGuard, Option<WorkerGuard>),
131 #[cfg(feature = "error_aggregation")]
132 pub errors_container: Arc<ErrorAggregationContainer>,
133 #[cfg(feature = "log_throttling")]
134 pub log_throttling_handle: TracingRateLimitLayer,
136}
137
138struct LoggingSubscriberParts {
140 subscriber: Box<dyn Subscriber + Send + Sync + 'static>,
141 reload_handle: LogReloadHandle,
142 log_guards: (WorkerGuard, Option<WorkerGuard>), #[cfg(feature = "error_aggregation")]
144 errors_container: Arc<ErrorAggregationContainer>,
145 #[cfg(feature = "log_throttling")]
146 log_throttling_handle: TracingRateLimitLayer,
147}
148
149fn build_logging_subscriber(config: LoggingConfig) -> eyre::Result<LoggingSubscriberParts> {
152 let global_env_filter = build_env_filter(config.level)?;
154 let (reloadable_global_filter, global_reload_handle) = reload::Layer::new(global_env_filter);
155
156 let (non_blocking_stdout, stdout_guard) = tracing_appender::non_blocking(std::io::stdout());
157
158 let stdout_layer = tracing_subscriber::fmt::layer()
159 .with_thread_names(true)
160 .with_line_number(true)
161 .with_writer(non_blocking_stdout);
162
163 #[cfg(feature = "error_aggregation")]
164 let error_aggregation_config = config.error_aggregation.clone();
165
166 #[cfg(feature = "log_throttling")]
167 let throttling_config = config.throttling_config.clone().unwrap_or_default();
168
169 #[cfg(feature = "log_throttling")]
170 let mut exemptions = vec!["tracing_throttle::infrastructure::layer".to_string()]; #[cfg(feature = "log_throttling")]
172 exemptions.extend(throttling_config.exemptions.unwrap_or_default());
173
174 #[cfg(feature = "log_throttling")]
175 let rate_limit_filter = TracingRateLimitLayer::builder()
176 .with_excluded_fields(throttling_config.excluded_fields.unwrap_or_default())
177 .with_exempt_targets(exemptions)
178 .with_active_emission(throttling_config.summary_emission_interval.is_some())
179 .with_summary_interval(
180 throttling_config
181 .summary_emission_interval
182 .unwrap_or(Duration::from_secs(5 * 60)),
183 )
184 .build()
185 .expect("Error building tracing rate limit layer");
186
187 #[cfg(feature = "log_throttling")]
188 if let Some(metrics_duration) = throttling_config.metrics_emission_interval {
189 let metrics = rate_limit_filter.metrics().clone();
190
191 nagoya::runtime::background().spawn(async move {
194 loop {
195 nagoya::sleep(metrics_duration).await;
196
197 let snapshot = metrics.snapshot();
198 tracing::info!(
199 events_allowed = snapshot.events_allowed,
200 events_suppressed = snapshot.events_suppressed,
201 suppression_rate = format!("{:.1}%", snapshot.suppression_rate() * 100.0),
202 "Rate limiting metrics"
203 );
204 }
205 });
206 }
207
208 #[cfg(feature = "log_throttling")]
209 let log_throttling_handle = rate_limit_filter.clone();
210
211 let (file_layer, file_guard) = match config.file_config {
212 None => (None, None),
213 Some(file_config) => {
214 let appender = RollingFileAppender::builder()
215 .rotation(file_config.rotation.unwrap_or(LogRotation::NEVER))
216 .filename_prefix(file_config.file_prefix.unwrap_or_else(|| {
217 chrono::Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true)
218 })) .filename_suffix("log")
220 .build(file_config.path)?;
221
222 let (non_blocking_appender, guard) = tracing_appender::non_blocking(appender);
223
224 let base_layer = tracing_subscriber::fmt::layer()
225 .with_thread_names(true)
226 .with_line_number(true)
227 .with_ansi(false)
228 .with_writer(non_blocking_appender);
229
230 #[cfg(feature = "log_throttling")]
231 let file_layer = base_layer.with_filter(rate_limit_filter);
232
233 #[cfg(not(feature = "log_throttling"))]
234 let file_layer = base_layer;
235
236 (Some(file_layer), Some(guard))
237 }
238 };
239
240 let reload_handle = LogReloadHandle(global_reload_handle);
241
242 if config.otel_config.enabled {
248 tracing::warn!(
249 target: "otel::setup",
250 endpoint = config.otel_config.endpoint.as_deref().unwrap_or("unset"),
251 "otel_config.enabled is true but OTLP export has been removed from endpoint-libs - \
252 no traces or logs will be forwarded to a collector"
253 );
254 }
255
256 let subscriber = registry().with(reloadable_global_filter);
260
261 let sinks = stdout_layer.and_then(file_layer);
263
264 #[cfg(feature = "error_aggregation")]
266 let (sinks, errors_container) = {
267 use crate::libs::log::error_aggregation::get_error_aggregation;
268 let (container, error_layer) = get_error_aggregation(error_aggregation_config);
269 (sinks.and_then(error_layer), container)
270 };
271
272 let subscriber = subscriber.with(sinks);
274
275 let subscriber: Box<dyn Subscriber + Send + Sync + 'static> = Box::new(subscriber);
280
281 Ok(LoggingSubscriberParts {
282 subscriber,
283 reload_handle,
284 log_guards: (stdout_guard, file_guard),
285 #[cfg(feature = "error_aggregation")]
286 errors_container,
287 #[cfg(feature = "log_throttling")]
288 log_throttling_handle,
289 })
290}
291
292pub fn setup_logging(config: LoggingConfig) -> eyre::Result<LogSetupReturn> {
311 use tracing_subscriber::util::SubscriberInitExt;
312
313 let parts = build_logging_subscriber(config)?;
314
315 parts.subscriber.init();
316
317 Ok(LogSetupReturn {
318 reload_handle: parts.reload_handle,
319 log_guards: parts.log_guards,
320 #[cfg(feature = "error_aggregation")]
321 errors_container: parts.errors_container,
322 #[cfg(feature = "log_throttling")]
323 log_throttling_handle: parts.log_throttling_handle,
324 })
325}
326
327pub type GlobalLogReloadHandle = Handle<EnvFilter, Registry>;
329
330#[derive(Clone)]
332pub struct LogReloadHandle(GlobalLogReloadHandle);
333
334impl std::fmt::Debug for LogReloadHandle {
335 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
336 f.debug_struct("LogReloadHandle")
337 .field("inner", &"Handle<EnvFilter, Registry>")
338 .finish()
339 }
340}
341
342impl LogReloadHandle {
343 pub fn set_log_level(&self, level: LogLevel) -> eyre::Result<()> {
346 match build_env_filter(level) {
347 Ok(filter) => match self.0.modify(|env_filter| *env_filter = filter) {
348 Ok(_) => Ok(()),
349 Err(error) => {
350 tracing::error!(
351 ?error,
352 "Error setting new global log filter. Ignoring reload attempt"
353 );
354 bail!("Error setting new global log filter: {error}. Ignoring reload attempt")
355 }
356 },
357 Err(error) => {
358 tracing::error!(
359 ?error,
360 "Error building new filter from given log level. Ignoring reload attempt"
361 );
362 bail!(
363 "Error building new filter from given log level: {error}. Ignoring reload attempt"
364 )
365 }
366 }
367 }
368}
369
370pub struct CustomEyreHandler {
374 default_handler: Box<dyn EyreHandler>,
375 location: Option<&'static std::panic::Location<'static>>,
376}
377
378impl CustomEyreHandler {
379 pub fn default_with_location_saving(
380 error: &(dyn std::error::Error + 'static),
381 ) -> Box<dyn EyreHandler> {
382 Box::new(Self {
383 default_handler: DefaultHandler::default_with(error),
384 location: None,
385 })
386 }
387
388 pub fn get_location(&self) -> &Option<&'static std::panic::Location<'static>> {
389 &self.location
390 }
391}
392
393impl EyreHandler for CustomEyreHandler {
394 fn display(
395 &self,
396 error: &(dyn std::error::Error + 'static),
397 f: &mut core::fmt::Formatter<'_>,
398 ) -> core::fmt::Result {
399 self.default_handler.display(error, f)
400 }
401
402 fn debug(
403 &self,
404 error: &(dyn std::error::Error + 'static),
405 f: &mut core::fmt::Formatter<'_>,
406 ) -> core::fmt::Result {
407 self.default_handler.debug(error, f)
408 }
409
410 fn track_caller(&mut self, location: &'static std::panic::Location<'static>) {
411 self.location = Some(location); self.default_handler.track_caller(location);
413 }
414}
415
416#[cfg(test)]
417pub struct LogSetupReturnTest {
418 _guard: tracing::subscriber::DefaultGuard,
419 reload_handle: LogReloadHandle,
420 #[allow(dead_code)]
421 log_guards: (WorkerGuard, Option<WorkerGuard>),
422 #[cfg(feature = "error_aggregation")]
423 #[allow(dead_code)]
424 errors_container: Arc<ErrorAggregationContainer>,
425}
426
427#[cfg(test)]
430pub fn setup_logging_test(config: LoggingConfig) -> eyre::Result<LogSetupReturnTest> {
431 let parts = build_logging_subscriber(config)?;
432
433 let guard = tracing::subscriber::set_default(parts.subscriber);
435
436 Ok(LogSetupReturnTest {
437 _guard: guard,
438 reload_handle: parts.reload_handle,
439 log_guards: parts.log_guards,
440 #[cfg(feature = "error_aggregation")]
441 errors_container: parts.errors_container,
442 })
443}
444
445#[cfg(test)]
446mod tests {
447 use super::*;
448 use std::fs;
449 use tracing::{debug, error, info};
450
451 #[cfg(feature = "error_aggregation")]
452 fn default_error_aggregation_config() -> ErrorAggregationConfig {
453 ErrorAggregationConfig {
454 limit: 100,
455 normalize: true,
456 }
457 }
458
459 #[test]
464 fn test_basic_logging_stdout() {
465 let config = LoggingConfig {
466 level: LogLevel::Info,
467 file_config: None,
468 otel_config: OtelConfig::default(),
469 #[cfg(feature = "error_aggregation")]
470 error_aggregation: default_error_aggregation_config(),
471 #[cfg(feature = "log_throttling")]
472 throttling_config: Some(LogThrottlingConfig::default()),
473 };
474
475 let _guard = setup_logging_test(config);
476 assert!(_guard.is_ok(), "Failed to setup logging");
477
478 info!("Test info message");
480 error!("Test error message");
481 }
482
483 #[test]
488 fn test_file_logging_comprehensive() {
489 let temp_dir = tempfile::tempdir().unwrap();
490
491 let config = LoggingConfig {
492 level: LogLevel::Info,
493 file_config: Some(FileLoggingConfig {
494 path: temp_dir.path().to_path_buf(),
495 file_prefix: Some("test".to_string()),
496 rotation: None,
497 }),
498 otel_config: OtelConfig::default(),
499 #[cfg(feature = "error_aggregation")]
500 error_aggregation: default_error_aggregation_config(),
501 #[cfg(feature = "log_throttling")]
502 throttling_config: Some(LogThrottlingConfig::default()),
503 };
504
505 let log_setup = setup_logging_test(config).unwrap();
506
507 info!("info_message_before_reload");
509 debug!("debug_message_before_reload");
510 error!("error_message");
511
512 std::thread::sleep(std::time::Duration::from_millis(100));
513
514 let result = log_setup.reload_handle.set_log_level(LogLevel::Debug);
516 assert!(result.is_ok());
517
518 info!("info_message_after_reload");
520 debug!("debug_message_after_reload");
521
522 std::thread::sleep(std::time::Duration::from_millis(100));
523 drop(log_setup);
524
525 std::thread::sleep(std::time::Duration::from_millis(100));
526
527 let log_files: Vec<_> = fs::read_dir(temp_dir.path()).unwrap().collect();
529 assert_eq!(log_files.len(), 1, "Expected exactly one log file");
530
531 let log_file = log_files[0].as_ref().unwrap();
532 let log_contents = fs::read_to_string(log_file.path()).unwrap();
533
534 assert!(log_contents.contains("info_message_before_reload"));
536 assert!(
537 !log_contents.contains("debug_message_before_reload"),
538 "Debug should not appear before reload"
539 );
540 assert!(log_contents.contains("error_message"));
541
542 assert!(log_contents.contains("info_message_after_reload"));
544
545 assert!(
547 log_contents.contains("debug_message_after_reload"),
548 "Debug should appear after reload to DEBUG level"
549 );
550 }
551
552 #[test]
569 fn enabling_otel_is_accepted_and_changes_nothing() {
570 let config = LoggingConfig {
571 level: LogLevel::Info,
572 file_config: None,
573 otel_config: OtelConfig {
574 enabled: true,
575 endpoint: Some("http://nonexistent.invalid:4317".to_string()),
576 ..OtelConfig::default()
577 },
578 #[cfg(feature = "error_aggregation")]
579 error_aggregation: default_error_aggregation_config(),
580 #[cfg(feature = "log_throttling")]
581 throttling_config: None,
582 };
583
584 let result = setup_logging_test(config);
585 assert!(
586 result.is_ok(),
587 "an otel_config asking for export must not fail setup"
588 );
589 }
590
591 #[test]
593 fn a_default_otel_config_sets_up_cleanly() {
594 let config = LoggingConfig {
595 level: LogLevel::Info,
596 file_config: None,
597 otel_config: OtelConfig::default(),
598 #[cfg(feature = "error_aggregation")]
599 error_aggregation: default_error_aggregation_config(),
600 #[cfg(feature = "log_throttling")]
601 throttling_config: None,
602 };
603
604 assert!(setup_logging_test(config).is_ok());
605 }
606
607 #[test]
610 fn test_file_logging_with_otel_attempted() {
611 let temp_dir = tempfile::tempdir().unwrap();
612
613 let config = LoggingConfig {
614 level: LogLevel::Info,
615 file_config: Some(FileLoggingConfig {
616 path: temp_dir.path().to_path_buf(),
617 file_prefix: Some("otel_test".to_string()),
618 rotation: None,
619 }),
620 otel_config: OtelConfig {
621 enabled: true,
622 endpoint: Some("http://localhost:4317".to_string()), ..OtelConfig::default()
624 },
625 #[cfg(feature = "error_aggregation")]
626 error_aggregation: default_error_aggregation_config(),
627 #[cfg(feature = "log_throttling")]
628 throttling_config: None,
629 };
630
631 let log_setup = setup_logging_test(config).unwrap();
632
633 info!("file_log_with_otel_attempted");
635 error!("error_log_with_otel_attempted");
636
637 std::thread::sleep(std::time::Duration::from_millis(100));
638 drop(log_setup);
639 std::thread::sleep(std::time::Duration::from_millis(100));
640
641 let log_files: Vec<_> = fs::read_dir(temp_dir.path()).unwrap().collect();
643 assert_eq!(log_files.len(), 1, "Expected exactly one log file");
644
645 let log_contents = fs::read_to_string(log_files[0].as_ref().unwrap().path()).unwrap();
646 assert!(log_contents.contains("file_log_with_otel_attempted"));
647 assert!(log_contents.contains("error_log_with_otel_attempted"));
648 }
649
650 #[test]
652 fn test_log_level_reload_with_otel_enabled() {
653 let temp_dir = tempfile::tempdir().unwrap();
654
655 let config = LoggingConfig {
656 level: LogLevel::Info,
657 file_config: Some(FileLoggingConfig {
658 path: temp_dir.path().to_path_buf(),
659 file_prefix: Some("reload_test".to_string()),
660 rotation: None,
661 }),
662 otel_config: OtelConfig {
663 enabled: true,
664 endpoint: Some("http://localhost:4317".to_string()),
665 ..OtelConfig::default()
666 },
667 #[cfg(feature = "error_aggregation")]
668 error_aggregation: default_error_aggregation_config(),
669 #[cfg(feature = "log_throttling")]
670 throttling_config: None,
671 };
672
673 let log_setup = setup_logging_test(config).unwrap();
674
675 info!("info_before_reload");
677 debug!("debug_before_reload"); std::thread::sleep(std::time::Duration::from_millis(100));
680
681 let result = log_setup.reload_handle.set_log_level(LogLevel::Debug);
683 assert!(
684 result.is_ok(),
685 "Log level reload should succeed with OTel enabled"
686 );
687
688 info!("info_after_reload");
690 debug!("debug_after_reload");
691
692 std::thread::sleep(std::time::Duration::from_millis(100));
693 drop(log_setup);
694 std::thread::sleep(std::time::Duration::from_millis(100));
695
696 let log_files: Vec<_> = fs::read_dir(temp_dir.path()).unwrap().collect();
697 let log_contents = fs::read_to_string(log_files[0].as_ref().unwrap().path()).unwrap();
698
699 assert!(log_contents.contains("info_before_reload"));
700 assert!(
701 !log_contents.contains("debug_before_reload"),
702 "Debug should not appear before reload"
703 );
704 assert!(log_contents.contains("info_after_reload"));
705 assert!(
706 log_contents.contains("debug_after_reload"),
707 "Debug should appear after reload"
708 );
709 }
710
711 #[test]
713 fn test_otel_config_fields() {
714 use std::collections::HashMap;
715
716 let config = OtelConfig {
717 enabled: true,
718 service_name: Some("test-service".to_string()),
719 endpoint: Some("http://collector:4317".to_string()),
720 headers: {
721 let mut h = HashMap::new();
722 h.insert("x-api-key".to_string(), "secret-key".to_string());
723 h
724 },
725 };
726
727 assert!(config.enabled);
728 assert_eq!(config.service_name, Some("test-service".to_string()));
729 assert_eq!(config.endpoint, Some("http://collector:4317".to_string()));
730 assert_eq!(
731 config.headers.get("x-api-key"),
732 Some(&"secret-key".to_string())
733 );
734
735 let default_config = OtelConfig::default();
737 assert!(!default_config.enabled);
738 assert!(default_config.service_name.is_none());
739 assert!(default_config.endpoint.is_none());
740 assert!(default_config.headers.is_empty());
741 }
742}