use std::path::PathBuf;
use nemo_relay::error::FlowError;
use nemo_relay::logging::{
DEFAULT_FILE_FLUSH_INTERVAL_MILLIS, DEFAULT_FILE_SINK_QUEUE_ENTRIES, FileLogRotationConfig,
FileLogSinkConfig, LogFormat, LogLevel, LogSinkConfig, LoggingConfig,
MAX_FILE_SINK_QUEUE_ENTRIES,
};
use serde::Deserialize;
use super::CliError;
#[derive(Debug, Clone, Default, Deserialize)]
#[serde(deny_unknown_fields)]
pub(super) struct FileLoggingConfig {
level: Option<String>,
stderr_format: Option<String>,
flush_interval_millis: Option<u64>,
#[serde(default)]
sinks: Vec<RawFileLogSinkConfig>,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(deny_unknown_fields)]
struct RawFileLogSinkConfig {
path: Option<PathBuf>,
level: Option<String>,
format: Option<String>,
queue_capacity: Option<usize>,
max_file_size_bytes: Option<u64>,
retained_files: Option<usize>,
}
pub(super) fn apply_file_logging_config(
logging: &mut LoggingConfig,
config: Option<FileLoggingConfig>,
) -> Result<(), CliError> {
let Some(config) = config else {
return Ok(());
};
if let Some(level) = config.level.as_deref() {
logging.level = LogLevel::parse(level).map_err(logging_parse_error)?;
}
if let Some(stderr_format) = config.stderr_format.as_deref() {
logging.stderr_format = LogFormat::parse(stderr_format).map_err(logging_parse_error)?;
}
logging.flush_interval_millis = config
.flush_interval_millis
.unwrap_or(DEFAULT_FILE_FLUSH_INTERVAL_MILLIS);
if !config.sinks.is_empty() {
let default_sink_level = logging.level;
logging.sinks = config
.sinks
.into_iter()
.map(|sink| parse_file_log_sink(sink, default_sink_level))
.collect::<Result<Vec<_>, _>>()?;
}
Ok(())
}
fn parse_file_log_sink(
config: RawFileLogSinkConfig,
default_level: LogLevel,
) -> Result<LogSinkConfig, CliError> {
let path = config
.path
.ok_or_else(|| CliError::Config("logging sink requires path".into()))?;
if path.as_os_str().is_empty() {
return Err(CliError::Config(
"logging sink path must not be empty".into(),
));
}
let level = match config.level.as_deref() {
Some(raw) => LogLevel::parse(raw).map_err(logging_parse_error)?,
None => default_level,
};
let format = match config.format.as_deref() {
Some(raw) => LogFormat::parse(raw).map_err(logging_parse_error)?,
None => LogFormat::Jsonl,
};
let queue_capacity = match config.queue_capacity {
Some(0) => {
return Err(CliError::Config(
"logging sink queue_capacity must be greater than 0".into(),
));
}
Some(capacity) if capacity > MAX_FILE_SINK_QUEUE_ENTRIES => {
return Err(CliError::Config(format!(
"logging sink queue_capacity {capacity} exceeds maximum \
{MAX_FILE_SINK_QUEUE_ENTRIES} entries per file sink"
)));
}
Some(capacity) => capacity,
None => DEFAULT_FILE_SINK_QUEUE_ENTRIES,
};
let rotation = match (config.max_file_size_bytes, config.retained_files) {
(None, None) => None,
(Some(max_file_size_bytes), Some(retained_files)) => Some(
FileLogRotationConfig::new(max_file_size_bytes, retained_files)
.map_err(logging_parse_error)?,
),
_ => {
return Err(CliError::Config(
"logging sink max_file_size_bytes and retained_files must be configured together"
.into(),
));
}
};
Ok(LogSinkConfig::File(FileLogSinkConfig {
path,
level,
format,
queue_capacity,
rotation,
}))
}
fn logging_parse_error(error: FlowError) -> CliError {
match error {
FlowError::InvalidArgument(message) => CliError::Config(message),
other => CliError::Flow(other),
}
}