@@ -798,6 +798,39 @@ pub fn server_main<E: Engine>(engine: E) -> Result<(), ControllerError> { |
798 | 798 | /// The time limit for the server to receive the first request from a new client. |
799 | 799 | const CLIENT_REQUEST_TIMEOUT: Duration = Duration::from_secs(180); |
800 | 800 |
|
| 801 | +/// Accumulates log messages until logging can be initialized. |
| 802 | +/// |
| 803 | +/// Logs the collected messages when dropped, so that they are not lost. |
| 804 | +#[derive(Default)] |
| 805 | +struct DeferredLogging(Vec<(Level, String)>); |
| 806 | + |
| 807 | +impl DeferredLogging { |
| 808 | + fn push(&mut self, level: Level, message: impl Into<String>) { |
| 809 | + self.0.push((level, message.into())); |
| 810 | + } |
| 811 | + |
| 812 | + /// Logs the queued messages, or writes them to `fallback` if there is no |
| 813 | + /// logger (as on an early return before the logger is initialized). |
| 814 | + fn flush(&mut self, fallback: &mut dyn std::io::Write) { |
| 815 | + let has_logger = tracing::dispatcher::get_default(|dispatch| { |
| 816 | + !dispatch.is::<tracing::subscriber::NoSubscriber>() |
| 817 | + }); |
| 818 | + for (level, message) in self.0.drain(..) { |
| 819 | + if has_logger { |
| 820 | + dyn_event!(level, "{message}"); |
| 821 | + } else { |
| 822 | + let _ = writeln!(fallback, "{message}"); |
| 823 | + } |
| 824 | + } |
| 825 | + } |
| 826 | +} |
| 827 | + |
| 828 | +impl Drop for DeferredLogging { |
| 829 | + fn drop(&mut self) { |
| 830 | + self.flush(&mut std::io::stderr()); |
| 831 | + } |
| 832 | +} |
| 833 | + |
801 | 834 | // We pass the engine as trait objects to make sure that this function |
802 | 835 | // doesn't have any generics and gets generated, along with the entire web server |
803 | 836 | // as part of this crate. This reduces the size of the generated code for |
@@ -808,6 +841,7 @@ pub fn run_server( |
808 | 841 | open_storage: OpenStorageFn, |
809 | 842 | program_loader: Option<Box<dyn ProgramLoader>>, |
810 | 843 | ) -> Result<(), ControllerError> { |
| 844 | + let mut deferred_logging = DeferredLogging::default(); |
811 | 845 | check_program_load_flag(&args, program_loader.is_some())?; |
812 | 846 | ensure_default_crypto_provider(); |
813 | 847 | enable_counting_panics(); |
@@ -824,11 +858,8 @@ pub fn run_server( |
824 | 858 | })? |
825 | 859 | .port(); |
826 | 860 |
|
827 | | - let config = parse_config(&args.config_file).inspect_err(|e| { |
828 | | - // Logging isn't initialized and we can't initialize it until we have |
829 | | - // the configuration, so just print the message to stderr for now. |
830 | | - eprintln!("{e}"); |
831 | | - })?; |
| 861 | + let config = parse_config_with_deferred_logging(&args.config_file, &mut deferred_logging) |
| 862 | + .inspect_err(|e| deferred_logging.push(Level::ERROR, e.to_string()))?; |
832 | 863 |
|
833 | 864 | // Initialize the logger by setting its filter and template. |
834 | 865 | let pipeline_label = config |
@@ -848,11 +879,16 @@ pub fn run_server( |
848 | 879 | .as_ref() |
849 | 880 | .and_then(|name| name.strip_prefix("pipeline-")) |
850 | 881 | .map(|id| id.to_string()); |
851 | | - init_pipeline_logging_with_id(pipeline_name, pipeline_id, get_env_filter(&config)) |
852 | | - .unwrap_or_else(|e| { |
853 | | - // This happens in unit tests when another test has initialized logging. |
854 | | - eprintln!("Failed to initialize logging: {e}.") |
855 | | - }); |
| 882 | + init_pipeline_logging_with_id( |
| 883 | + pipeline_name, |
| 884 | + pipeline_id, |
| 885 | + get_env_filter(&config, &mut deferred_logging), |
| 886 | + ) |
| 887 | + .unwrap_or_else(|e| { |
| 888 | + // This happens in unit tests when another test has initialized logging. |
| 889 | + deferred_logging.push(Level::WARN, format!("Failed to initialize logging: {e}.")); |
| 890 | + }); |
| 891 | + drop(deferred_logging); |
856 | 892 | if let Some(provider) = rustls::crypto::CryptoProvider::get_default() { |
857 | 893 | observability::fips::log_rustls_provider_fips_status( |
858 | 894 | "startup default provider", |
@@ -1272,8 +1308,14 @@ fn check_program_load_flag(args: &ServerArgs, can_load: bool) -> Result<(), Cont |
1272 | 1308 | Ok(()) |
1273 | 1309 | } |
1274 | 1310 |
|
1275 | | -fn parse_config(config_file: impl AsRef<Path>) -> Result<PipelineConfig, ControllerError> { |
1276 | | - fn read_config_file_as_string(path: &Path) -> Result<String, ControllerError> { |
| 1311 | +fn parse_config_with_deferred_logging( |
| 1312 | + config_file: impl AsRef<Path>, |
| 1313 | + deferred_logging: &mut DeferredLogging, |
| 1314 | +) -> Result<PipelineConfig, ControllerError> { |
| 1315 | + fn read_config_file_as_string( |
| 1316 | + path: &Path, |
| 1317 | + deferred_logging: &mut DeferredLogging, |
| 1318 | + ) -> Result<String, ControllerError> { |
1277 | 1319 | let string = std::fs::read(path).map_err(|e| { |
1278 | 1320 | ControllerError::io_error( |
1279 | 1321 | format!("reading configuration file '{}'", path.display()), |
@@ -1288,14 +1330,16 @@ fn parse_config(config_file: impl AsRef<Path>) -> Result<PipelineConfig, Control |
1288 | 1330 | )) |
1289 | 1331 | })?; |
1290 | 1332 |
|
1291 | | - // Still running without logger here. |
1292 | | - eprintln!("Pipeline configuration read from {}.", path.display()); |
| 1333 | + deferred_logging.push( |
| 1334 | + Level::INFO, |
| 1335 | + format!("Pipeline configuration read from {}.", path.display()), |
| 1336 | + ); |
1293 | 1337 |
|
1294 | 1338 | Ok(string) |
1295 | 1339 | } |
1296 | 1340 |
|
1297 | 1341 | let path = config_file.as_ref(); |
1298 | | - let string = read_config_file_as_string(path)?; |
| 1342 | + let string = read_config_file_as_string(path, deferred_logging)?; |
1299 | 1343 | let mut config: PipelineConfig = serde_json::from_str(&string).map_err(|e| { |
1300 | 1344 | ControllerError::pipeline_config_parse_error(&format!( |
1301 | 1345 | "configuration file '{}' must be JSON (YAML is no longer read): {e}", |
@@ -1305,10 +1349,12 @@ fn parse_config(config_file: impl AsRef<Path>) -> Result<PipelineConfig, Control |
1305 | 1349 |
|
1306 | 1350 | if let Ok(value) = serde_json::from_str(&string) { |
1307 | 1351 | for setting in ignored_settings::<PipelineConfig>(&value) { |
1308 | | - // The logger is not running yet. |
1309 | | - eprintln!( |
1310 | | - "Ignoring pipeline configuration setting '{setting}', which this version \ |
| 1352 | + deferred_logging.push( |
| 1353 | + Level::WARN, |
| 1354 | + format!( |
| 1355 | + "Ignoring pipeline configuration setting '{setting}', which this version \ |
1311 | 1356 | of the pipeline does not recognize (it may be misspelled, or newer than this pipeline)." |
| 1357 | + ), |
1312 | 1358 | ); |
1313 | 1359 | } |
1314 | 1360 | } |
@@ -1319,22 +1365,35 @@ of the pipeline does not recognize (it may be misspelled, or newer than this pip |
1319 | 1365 | && let Some(available_mb) = observability::system::total_memory_megabyte() |
1320 | 1366 | { |
1321 | 1367 | let available = ConfigSize::from_mb(available_mb); |
1322 | | - // The logger is not running yet. |
1323 | | - eprintln!( |
1324 | | - "No memory limit configured ('max_rss' or 'resources.memory_max'); \ |
| 1368 | + deferred_logging.push( |
| 1369 | + Level::INFO, |
| 1370 | + format!( |
| 1371 | + "No memory limit configured ('max_rss' or 'resources.memory_max'); \ |
1325 | 1372 | using this host's available memory: 'resources.memory_max' = {available}." |
| 1373 | + ), |
1326 | 1374 | ); |
1327 | 1375 | config.global.resources.memory_max = Some(available); |
1328 | 1376 | } |
1329 | 1377 |
|
1330 | | - eprintln!( |
1331 | | - "Pipeline configuration loaded successfully: {}", |
1332 | | - config.display_summary() |
| 1378 | + deferred_logging.push( |
| 1379 | + Level::INFO, |
| 1380 | + format!( |
| 1381 | + "Pipeline configuration loaded successfully: {}", |
| 1382 | + config.display_summary() |
| 1383 | + ), |
1333 | 1384 | ); |
1334 | 1385 |
|
1335 | 1386 | Ok(config) |
1336 | 1387 | } |
1337 | 1388 |
|
| 1389 | +#[cfg(any(test, feature = "testing"))] |
| 1390 | +fn parse_config(config_file: impl AsRef<Path>) -> Result<PipelineConfig, ControllerError> { |
| 1391 | + let mut deferred_logging = DeferredLogging::default(); |
| 1392 | + let result = parse_config_with_deferred_logging(config_file, &mut deferred_logging); |
| 1393 | + deferred_logging.0.clear(); |
| 1394 | + result |
| 1395 | +} |
| 1396 | + |
1338 | 1397 | // Initialization thread function. |
1339 | 1398 | fn bootstrap( |
1340 | 1399 | builder: ControllerBuilder, |
@@ -1415,7 +1474,7 @@ fn error_handler( |
1415 | 1474 | } |
1416 | 1475 |
|
1417 | 1476 | /// Get the log filtering configuration. |
1418 | | -fn get_env_filter(config: &PipelineConfig) -> EnvFilter { |
| 1477 | +fn get_env_filter(config: &PipelineConfig, deferred_logging: &mut DeferredLogging) -> EnvFilter { |
1419 | 1478 | // The `RUST_LOG` environment variable takes precedence. It isn't usually |
1420 | 1479 | // set. |
1421 | 1480 | if let Ok(env_filter) = EnvFilter::try_from_default_env() { |
@@ -1427,8 +1486,10 @@ fn get_env_filter(config: &PipelineConfig) -> EnvFilter { |
1427 | 1486 | match EnvFilter::try_new(dirs) { |
1428 | 1487 | Ok(env_filter) => return env_filter, |
1429 | 1488 | Err(error) => { |
1430 | | - // Write directly to stderr because logging isn't set up yet. |
1431 | | - eprintln!("Invalid pipeline `logging` configuration ({error}): {dirs}") |
| 1489 | + deferred_logging.push( |
| 1490 | + Level::ERROR, |
| 1491 | + format!("Invalid pipeline `logging` configuration ({error}): {dirs}"), |
| 1492 | + ); |
1432 | 1493 | } |
1433 | 1494 | } |
1434 | 1495 | } |
@@ -5378,7 +5439,10 @@ fn is_false(value: &bool) -> bool { |
5378 | 5439 | #[cfg(test)] |
5379 | 5440 | mod test_status { |
5380 | 5441 | use super::test_http_helpers::json_config_file; |
5381 | | - use super::{ServerState, error_handler, get_status, parse_config, terminated_status}; |
| 5442 | + use super::{ |
| 5443 | + DeferredLogging, ServerState, error_handler, get_status, parse_config, |
| 5444 | + parse_config_with_deferred_logging, terminated_status, |
| 5445 | + }; |
5382 | 5446 | use crate::{ |
5383 | 5447 | ControllerError, |
5384 | 5448 | server::{InitializationState, PipelinePhase}, |
@@ -5388,7 +5452,11 @@ mod test_status { |
5388 | 5452 | BootstrapConfig, BootstrapPolicy, RuntimeDesiredStatus, RuntimeStatus, |
5389 | 5453 | }; |
5390 | 5454 | use serde_json::json; |
5391 | | - use std::{io::Write, sync::Arc}; |
| 5455 | + use std::{ |
| 5456 | + io::Write, |
| 5457 | + sync::{Arc, Mutex}, |
| 5458 | + }; |
| 5459 | + use tracing::{Level, subscriber::NoSubscriber}; |
5392 | 5460 | use uuid::Uuid; |
5393 | 5461 |
|
5394 | 5462 | /// The configuration file is read as JSON whatever its name: the YAML |
@@ -5411,6 +5479,101 @@ mod test_status { |
5411 | 5479 | assert_eq!(config.name.as_deref(), Some("test")); |
5412 | 5480 | } |
5413 | 5481 |
|
| 5482 | + /// A `Write` that tests can read back after a subscriber writes to it. |
| 5483 | + #[derive(Clone, Default)] |
| 5484 | + struct SharedBuffer(Arc<Mutex<Vec<u8>>>); |
| 5485 | + |
| 5486 | + impl Write for SharedBuffer { |
| 5487 | + fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> { |
| 5488 | + self.0.lock().unwrap().write(buf) |
| 5489 | + } |
| 5490 | + fn flush(&mut self) -> std::io::Result<()> { |
| 5491 | + Ok(()) |
| 5492 | + } |
| 5493 | + } |
| 5494 | + |
| 5495 | + impl SharedBuffer { |
| 5496 | + fn contents(&self) -> String { |
| 5497 | + String::from_utf8(self.0.lock().unwrap().clone()).unwrap() |
| 5498 | + } |
| 5499 | + } |
| 5500 | + |
| 5501 | + /// With a logger, queued messages go to it, in order and at their levels, |
| 5502 | + /// and nothing goes to the fallback. |
| 5503 | + #[test] |
| 5504 | + fn deferred_logging_uses_logger() { |
| 5505 | + let log = SharedBuffer::default(); |
| 5506 | + let subscriber = tracing_subscriber::fmt() |
| 5507 | + .with_writer({ |
| 5508 | + let log = log.clone(); |
| 5509 | + move || log.clone() |
| 5510 | + }) |
| 5511 | + .with_ansi(false) |
| 5512 | + .with_max_level(Level::INFO) |
| 5513 | + .finish(); |
| 5514 | + let mut fallback = Vec::new(); |
| 5515 | + tracing::subscriber::with_default(subscriber, || { |
| 5516 | + let mut deferred_logging = DeferredLogging::default(); |
| 5517 | + deferred_logging.push(Level::INFO, "first message"); |
| 5518 | + deferred_logging.push(Level::WARN, "second message"); |
| 5519 | + deferred_logging.push(Level::DEBUG, "filtered message"); |
| 5520 | + deferred_logging.flush(&mut fallback); |
| 5521 | + }); |
| 5522 | + let log = log.contents(); |
| 5523 | + let lines = log.lines().collect::<Vec<_>>(); |
| 5524 | + assert_eq!(lines.len(), 2, "{log}"); |
| 5525 | + assert!( |
| 5526 | + lines[0].contains("INFO") && lines[0].contains("first message"), |
| 5527 | + "{log}" |
| 5528 | + ); |
| 5529 | + assert!( |
| 5530 | + lines[1].contains("WARN") && lines[1].contains("second message"), |
| 5531 | + "{log}" |
| 5532 | + ); |
| 5533 | + assert!(fallback.is_empty()); |
| 5534 | + } |
| 5535 | + |
| 5536 | + /// Without a logger, as on an early return from `run_server`, queued |
| 5537 | + /// messages go to the fallback instead of being lost. |
| 5538 | + #[test] |
| 5539 | + fn deferred_logging_falls_back_without_logger() { |
| 5540 | + let mut fallback = Vec::new(); |
| 5541 | + tracing::subscriber::with_default(NoSubscriber::default(), || { |
| 5542 | + let mut deferred_logging = DeferredLogging::default(); |
| 5543 | + deferred_logging.push(Level::INFO, "first message"); |
| 5544 | + deferred_logging.push(Level::ERROR, "second message"); |
| 5545 | + deferred_logging.flush(&mut fallback); |
| 5546 | + }); |
| 5547 | + assert_eq!( |
| 5548 | + String::from_utf8(fallback).unwrap(), |
| 5549 | + "first message\nsecond message\n" |
| 5550 | + ); |
| 5551 | + } |
| 5552 | + |
| 5553 | + /// An unknown configuration setting queues a warning that names it. |
| 5554 | + #[test] |
| 5555 | + fn parse_config_warns_about_unknown_settings() { |
| 5556 | + let mut deferred_logging = DeferredLogging::default(); |
| 5557 | + parse_config_with_deferred_logging( |
| 5558 | + json_config_file(&json!({"name": "test", "no_such_setting": 1})).path(), |
| 5559 | + &mut deferred_logging, |
| 5560 | + ) |
| 5561 | + .unwrap(); |
| 5562 | + let warnings = std::mem::take(&mut deferred_logging.0) |
| 5563 | + .into_iter() |
| 5564 | + .filter(|(level, _)| *level == Level::WARN) |
| 5565 | + .map(|(_, message)| message) |
| 5566 | + .collect::<Vec<_>>(); |
| 5567 | + assert_eq!( |
| 5568 | + warnings, |
| 5569 | + [ |
| 5570 | + "Ignoring pipeline configuration setting 'no_such_setting', which this version \ |
| 5571 | + of the pipeline does not recognize (it may be misspelled, or newer than \ |
| 5572 | + this pipeline)." |
| 5573 | + ] |
| 5574 | + ); |
| 5575 | + } |
| 5576 | + |
5414 | 5577 | /// Without a memory limit, the pipeline sets `memory_max`, not `max_rss`, |
5415 | 5578 | /// from the available memory, so that a resize can change the budget. |
5416 | 5579 | #[test] |
|