@@ -795,6 +795,39 @@ pub fn server_main<E: Engine>(engine: E) -> Result<(), ControllerError> { |
795 | 795 | }) |
796 | 796 | } |
797 | 797 |
|
| 798 | +/// Accumulates log messages until logging can be initialized. |
| 799 | +/// |
| 800 | +/// Logs the collected messages when dropped, so that they are not lost. |
| 801 | +#[derive(Default)] |
| 802 | +struct DeferredLogging(Vec<(Level, String)>); |
| 803 | + |
| 804 | +impl DeferredLogging { |
| 805 | + fn push(&mut self, level: Level, message: impl Into<String>) { |
| 806 | + self.0.push((level, message.into())); |
| 807 | + } |
| 808 | + |
| 809 | + /// Logs the queued messages, or writes them to `fallback` if there is no |
| 810 | + /// logger (as on an early return before the logger is initialized). |
| 811 | + fn flush(&mut self, fallback: &mut dyn std::io::Write) { |
| 812 | + let has_logger = tracing::dispatcher::get_default(|dispatch| { |
| 813 | + !dispatch.is::<tracing::subscriber::NoSubscriber>() |
| 814 | + }); |
| 815 | + for (level, message) in self.0.drain(..) { |
| 816 | + if has_logger { |
| 817 | + dyn_event!(level, "{message}"); |
| 818 | + } else { |
| 819 | + let _ = writeln!(fallback, "{message}"); |
| 820 | + } |
| 821 | + } |
| 822 | + } |
| 823 | +} |
| 824 | + |
| 825 | +impl Drop for DeferredLogging { |
| 826 | + fn drop(&mut self) { |
| 827 | + self.flush(&mut std::io::stderr()); |
| 828 | + } |
| 829 | +} |
| 830 | + |
798 | 831 | // We pass the engine as trait objects to make sure that this function |
799 | 832 | // doesn't have any generics and gets generated, along with the entire web server |
800 | 833 | // as part of this crate. This reduces the size of the generated code for |
@@ -805,6 +838,7 @@ pub fn run_server( |
805 | 838 | open_storage: OpenStorageFn, |
806 | 839 | program_loader: Option<Box<dyn ProgramLoader>>, |
807 | 840 | ) -> Result<(), ControllerError> { |
| 841 | + let mut deferred_logging = DeferredLogging::default(); |
808 | 842 | check_program_load_flag(&args, program_loader.is_some())?; |
809 | 843 | ensure_default_crypto_provider(); |
810 | 844 | enable_counting_panics(); |
@@ -821,11 +855,8 @@ pub fn run_server( |
821 | 855 | })? |
822 | 856 | .port(); |
823 | 857 |
|
824 | | - let config = parse_config(&args.config_file).inspect_err(|e| { |
825 | | - // Logging isn't initialized and we can't initialize it until we have |
826 | | - // the configuration, so just print the message to stderr for now. |
827 | | - eprintln!("{e}"); |
828 | | - })?; |
| 858 | + let config = parse_config_with_deferred_logging(&args.config_file, &mut deferred_logging) |
| 859 | + .inspect_err(|e| deferred_logging.push(Level::ERROR, e.to_string()))?; |
829 | 860 |
|
830 | 861 | // Initialize the logger by setting its filter and template. |
831 | 862 | let pipeline_label = config |
@@ -845,11 +876,16 @@ pub fn run_server( |
845 | 876 | .as_ref() |
846 | 877 | .and_then(|name| name.strip_prefix("pipeline-")) |
847 | 878 | .map(|id| id.to_string()); |
848 | | - init_pipeline_logging_with_id(pipeline_name, pipeline_id, get_env_filter(&config)) |
849 | | - .unwrap_or_else(|e| { |
850 | | - // This happens in unit tests when another test has initialized logging. |
851 | | - eprintln!("Failed to initialize logging: {e}.") |
852 | | - }); |
| 879 | + init_pipeline_logging_with_id( |
| 880 | + pipeline_name, |
| 881 | + pipeline_id, |
| 882 | + get_env_filter(&config, &mut deferred_logging), |
| 883 | + ) |
| 884 | + .unwrap_or_else(|e| { |
| 885 | + // This happens in unit tests when another test has initialized logging. |
| 886 | + deferred_logging.push(Level::WARN, format!("Failed to initialize logging: {e}.")); |
| 887 | + }); |
| 888 | + drop(deferred_logging); |
853 | 889 | if let Some(provider) = rustls::crypto::CryptoProvider::get_default() { |
854 | 890 | observability::fips::log_rustls_provider_fips_status( |
855 | 891 | "startup default provider", |
@@ -1269,8 +1305,14 @@ fn check_program_load_flag(args: &ServerArgs, can_load: bool) -> Result<(), Cont |
1269 | 1305 | Ok(()) |
1270 | 1306 | } |
1271 | 1307 |
|
1272 | | -fn parse_config(config_file: impl AsRef<Path>) -> Result<PipelineConfig, ControllerError> { |
1273 | | - fn read_config_file_as_string(path: &Path) -> Result<String, ControllerError> { |
| 1308 | +fn parse_config_with_deferred_logging( |
| 1309 | + config_file: impl AsRef<Path>, |
| 1310 | + deferred_logging: &mut DeferredLogging, |
| 1311 | +) -> Result<PipelineConfig, ControllerError> { |
| 1312 | + fn read_config_file_as_string( |
| 1313 | + path: &Path, |
| 1314 | + deferred_logging: &mut DeferredLogging, |
| 1315 | + ) -> Result<String, ControllerError> { |
1274 | 1316 | let string = std::fs::read(path).map_err(|e| { |
1275 | 1317 | ControllerError::io_error( |
1276 | 1318 | format!("reading configuration file '{}'", path.display()), |
@@ -1285,14 +1327,16 @@ fn parse_config(config_file: impl AsRef<Path>) -> Result<PipelineConfig, Control |
1285 | 1327 | )) |
1286 | 1328 | })?; |
1287 | 1329 |
|
1288 | | - // Still running without logger here. |
1289 | | - eprintln!("Pipeline configuration read from {}.", path.display()); |
| 1330 | + deferred_logging.push( |
| 1331 | + Level::INFO, |
| 1332 | + format!("Pipeline configuration read from {}.", path.display()), |
| 1333 | + ); |
1290 | 1334 |
|
1291 | 1335 | Ok(string) |
1292 | 1336 | } |
1293 | 1337 |
|
1294 | 1338 | let path = config_file.as_ref(); |
1295 | | - let string = read_config_file_as_string(path)?; |
| 1339 | + let string = read_config_file_as_string(path, deferred_logging)?; |
1296 | 1340 | let mut config: PipelineConfig = serde_json::from_str(&string).map_err(|e| { |
1297 | 1341 | ControllerError::pipeline_config_parse_error(&format!( |
1298 | 1342 | "configuration file '{}' must be JSON (YAML is no longer read): {e}", |
@@ -1302,10 +1346,12 @@ fn parse_config(config_file: impl AsRef<Path>) -> Result<PipelineConfig, Control |
1302 | 1346 |
|
1303 | 1347 | if let Ok(value) = serde_json::from_str(&string) { |
1304 | 1348 | for setting in ignored_settings::<PipelineConfig>(&value) { |
1305 | | - // The logger is not running yet. |
1306 | | - eprintln!( |
1307 | | - "Ignoring pipeline configuration setting '{setting}', which this version \ |
| 1349 | + deferred_logging.push( |
| 1350 | + Level::WARN, |
| 1351 | + format!( |
| 1352 | + "Ignoring pipeline configuration setting '{setting}', which this version \ |
1308 | 1353 | of the pipeline does not recognize (it may be misspelled, or newer than this pipeline)." |
| 1354 | + ), |
1309 | 1355 | ); |
1310 | 1356 | } |
1311 | 1357 | } |
@@ -1316,22 +1362,35 @@ of the pipeline does not recognize (it may be misspelled, or newer than this pip |
1316 | 1362 | && let Some(available_mb) = observability::system::total_memory_megabyte() |
1317 | 1363 | { |
1318 | 1364 | let available = ConfigSize::from_mb(available_mb); |
1319 | | - // The logger is not running yet. |
1320 | | - eprintln!( |
1321 | | - "No memory limit configured ('max_rss' or 'resources.memory_max'); \ |
| 1365 | + deferred_logging.push( |
| 1366 | + Level::INFO, |
| 1367 | + format!( |
| 1368 | + "No memory limit configured ('max_rss' or 'resources.memory_max'); \ |
1322 | 1369 | using this host's available memory: 'resources.memory_max' = {available}." |
| 1370 | + ), |
1323 | 1371 | ); |
1324 | 1372 | config.global.resources.memory_max = Some(available); |
1325 | 1373 | } |
1326 | 1374 |
|
1327 | | - eprintln!( |
1328 | | - "Pipeline configuration loaded successfully: {}", |
1329 | | - config.display_summary() |
| 1375 | + deferred_logging.push( |
| 1376 | + Level::INFO, |
| 1377 | + format!( |
| 1378 | + "Pipeline configuration loaded successfully: {}", |
| 1379 | + config.display_summary() |
| 1380 | + ), |
1330 | 1381 | ); |
1331 | 1382 |
|
1332 | 1383 | Ok(config) |
1333 | 1384 | } |
1334 | 1385 |
|
| 1386 | +#[cfg(test)] |
| 1387 | +fn parse_config(config_file: impl AsRef<Path>) -> Result<PipelineConfig, ControllerError> { |
| 1388 | + let mut deferred_logging = DeferredLogging::default(); |
| 1389 | + let result = parse_config_with_deferred_logging(config_file, &mut deferred_logging); |
| 1390 | + deferred_logging.0.clear(); |
| 1391 | + result |
| 1392 | +} |
| 1393 | + |
1335 | 1394 | // Initialization thread function. |
1336 | 1395 | fn bootstrap( |
1337 | 1396 | builder: ControllerBuilder, |
@@ -1412,7 +1471,7 @@ fn error_handler( |
1412 | 1471 | } |
1413 | 1472 |
|
1414 | 1473 | /// Get the log filtering configuration. |
1415 | | -fn get_env_filter(config: &PipelineConfig) -> EnvFilter { |
| 1474 | +fn get_env_filter(config: &PipelineConfig, deferred_logging: &mut DeferredLogging) -> EnvFilter { |
1416 | 1475 | // The `RUST_LOG` environment variable takes precedence. It isn't usually |
1417 | 1476 | // set. |
1418 | 1477 | if let Ok(env_filter) = EnvFilter::try_from_default_env() { |
@@ -1424,8 +1483,10 @@ fn get_env_filter(config: &PipelineConfig) -> EnvFilter { |
1424 | 1483 | match EnvFilter::try_new(dirs) { |
1425 | 1484 | Ok(env_filter) => return env_filter, |
1426 | 1485 | Err(error) => { |
1427 | | - // Write directly to stderr because logging isn't set up yet. |
1428 | | - eprintln!("Invalid pipeline `logging` configuration ({error}): {dirs}") |
| 1486 | + deferred_logging.push( |
| 1487 | + Level::ERROR, |
| 1488 | + format!("Invalid pipeline `logging` configuration ({error}): {dirs}"), |
| 1489 | + ); |
1429 | 1490 | } |
1430 | 1491 | } |
1431 | 1492 | } |
@@ -5362,7 +5423,10 @@ fn is_false(value: &bool) -> bool { |
5362 | 5423 | #[cfg(test)] |
5363 | 5424 | mod test_status { |
5364 | 5425 | use super::test_http_helpers::json_config_file; |
5365 | | - use super::{ServerState, error_handler, get_status, parse_config, terminated_status}; |
| 5426 | + use super::{ |
| 5427 | + DeferredLogging, ServerState, error_handler, get_status, parse_config, |
| 5428 | + parse_config_with_deferred_logging, terminated_status, |
| 5429 | + }; |
5366 | 5430 | use crate::{ |
5367 | 5431 | ControllerError, |
5368 | 5432 | server::{InitializationState, PipelinePhase}, |
@@ -5372,7 +5436,11 @@ mod test_status { |
5372 | 5436 | BootstrapConfig, BootstrapPolicy, RuntimeDesiredStatus, RuntimeStatus, |
5373 | 5437 | }; |
5374 | 5438 | use serde_json::json; |
5375 | | - use std::{io::Write, sync::Arc}; |
| 5439 | + use std::{ |
| 5440 | + io::Write, |
| 5441 | + sync::{Arc, Mutex}, |
| 5442 | + }; |
| 5443 | + use tracing::{Level, subscriber::NoSubscriber}; |
5376 | 5444 | use uuid::Uuid; |
5377 | 5445 |
|
5378 | 5446 | /// The configuration file is read as JSON whatever its name: the YAML |
@@ -5395,6 +5463,101 @@ mod test_status { |
5395 | 5463 | assert_eq!(config.name.as_deref(), Some("test")); |
5396 | 5464 | } |
5397 | 5465 |
|
| 5466 | + /// A `Write` that tests can read back after a subscriber writes to it. |
| 5467 | + #[derive(Clone, Default)] |
| 5468 | + struct SharedBuffer(Arc<Mutex<Vec<u8>>>); |
| 5469 | + |
| 5470 | + impl Write for SharedBuffer { |
| 5471 | + fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> { |
| 5472 | + self.0.lock().unwrap().write(buf) |
| 5473 | + } |
| 5474 | + fn flush(&mut self) -> std::io::Result<()> { |
| 5475 | + Ok(()) |
| 5476 | + } |
| 5477 | + } |
| 5478 | + |
| 5479 | + impl SharedBuffer { |
| 5480 | + fn contents(&self) -> String { |
| 5481 | + String::from_utf8(self.0.lock().unwrap().clone()).unwrap() |
| 5482 | + } |
| 5483 | + } |
| 5484 | + |
| 5485 | + /// With a logger, queued messages go to it, in order and at their levels, |
| 5486 | + /// and nothing goes to the fallback. |
| 5487 | + #[test] |
| 5488 | + fn deferred_logging_uses_logger() { |
| 5489 | + let log = SharedBuffer::default(); |
| 5490 | + let subscriber = tracing_subscriber::fmt() |
| 5491 | + .with_writer({ |
| 5492 | + let log = log.clone(); |
| 5493 | + move || log.clone() |
| 5494 | + }) |
| 5495 | + .with_ansi(false) |
| 5496 | + .with_max_level(Level::INFO) |
| 5497 | + .finish(); |
| 5498 | + let mut fallback = Vec::new(); |
| 5499 | + tracing::subscriber::with_default(subscriber, || { |
| 5500 | + let mut deferred_logging = DeferredLogging::default(); |
| 5501 | + deferred_logging.push(Level::INFO, "first message"); |
| 5502 | + deferred_logging.push(Level::WARN, "second message"); |
| 5503 | + deferred_logging.push(Level::DEBUG, "filtered message"); |
| 5504 | + deferred_logging.flush(&mut fallback); |
| 5505 | + }); |
| 5506 | + let log = log.contents(); |
| 5507 | + let lines = log.lines().collect::<Vec<_>>(); |
| 5508 | + assert_eq!(lines.len(), 2, "{log}"); |
| 5509 | + assert!( |
| 5510 | + lines[0].contains("INFO") && lines[0].contains("first message"), |
| 5511 | + "{log}" |
| 5512 | + ); |
| 5513 | + assert!( |
| 5514 | + lines[1].contains("WARN") && lines[1].contains("second message"), |
| 5515 | + "{log}" |
| 5516 | + ); |
| 5517 | + assert!(fallback.is_empty()); |
| 5518 | + } |
| 5519 | + |
| 5520 | + /// Without a logger, as on an early return from `run_server`, queued |
| 5521 | + /// messages go to the fallback instead of being lost. |
| 5522 | + #[test] |
| 5523 | + fn deferred_logging_falls_back_without_logger() { |
| 5524 | + let mut fallback = Vec::new(); |
| 5525 | + tracing::subscriber::with_default(NoSubscriber::default(), || { |
| 5526 | + let mut deferred_logging = DeferredLogging::default(); |
| 5527 | + deferred_logging.push(Level::INFO, "first message"); |
| 5528 | + deferred_logging.push(Level::ERROR, "second message"); |
| 5529 | + deferred_logging.flush(&mut fallback); |
| 5530 | + }); |
| 5531 | + assert_eq!( |
| 5532 | + String::from_utf8(fallback).unwrap(), |
| 5533 | + "first message\nsecond message\n" |
| 5534 | + ); |
| 5535 | + } |
| 5536 | + |
| 5537 | + /// An unknown configuration setting queues a warning that names it. |
| 5538 | + #[test] |
| 5539 | + fn parse_config_warns_about_unknown_settings() { |
| 5540 | + let mut deferred_logging = DeferredLogging::default(); |
| 5541 | + parse_config_with_deferred_logging( |
| 5542 | + json_config_file(&json!({"name": "test", "no_such_setting": 1})).path(), |
| 5543 | + &mut deferred_logging, |
| 5544 | + ) |
| 5545 | + .unwrap(); |
| 5546 | + let warnings = std::mem::take(&mut deferred_logging.0) |
| 5547 | + .into_iter() |
| 5548 | + .filter(|(level, _)| *level == Level::WARN) |
| 5549 | + .map(|(_, message)| message) |
| 5550 | + .collect::<Vec<_>>(); |
| 5551 | + assert_eq!( |
| 5552 | + warnings, |
| 5553 | + [ |
| 5554 | + "Ignoring pipeline configuration setting 'no_such_setting', which this version \ |
| 5555 | + of the pipeline does not recognize (it may be misspelled, or newer than \ |
| 5556 | + this pipeline)." |
| 5557 | + ] |
| 5558 | + ); |
| 5559 | + } |
| 5560 | + |
5398 | 5561 | /// Without a memory limit, the pipeline sets `memory_max`, not `max_rss`, |
5399 | 5562 | /// from the available memory, so that a resize can change the budget. |
5400 | 5563 | #[test] |
|