diff --git a/geoengine/operators/src/bin/gdalsource_process_main.rs b/geoengine/operators/src/bin/gdalsource_process_main.rs index 0415e1227..9cd300769 100644 --- a/geoengine/operators/src/bin/gdalsource_process_main.rs +++ b/geoengine/operators/src/bin/gdalsource_process_main.rs @@ -108,13 +108,15 @@ fn init_subscriber( logging_config: &WorkerLoggingConfig, open_telemetry_config: &OpenTelemetryConfig, token: &str, -) -> (Option, Option) { +) -> (Option, WorkerGuard, Option) { + // we use the non-blocking wrapper around stderr so that we don't block the worker on logging. The guard is dropped at the end of main to flush any remaining logs. + let (non_blocking_writer, stderr_guard) = tracing_appender::non_blocking(std::io::stderr()); let stderr_layer = tracing_subscriber::fmt::layer() .pretty() .with_file(false) .with_target(true) .with_ansi(true) - .with_writer(std::io::stderr) + .with_writer(non_blocking_writer) .with_filter(EnvFilter::new(&logging_config.log_spec)); let (file_layer, file_guard) = if logging_config.log_to_file { @@ -162,14 +164,14 @@ fn init_subscriber( .with(opentelemetry) .init(); - return (Some(provider), file_guard); + return (Some(provider), stderr_guard, file_guard); } tracing_subscriber::registry() .with(stderr_layer) .with(file_layer) .init(); - (None, file_guard) + (None, stderr_guard, file_guard) } // We use a custom formatter because there are still format flags within spans even when @@ -227,7 +229,7 @@ fn main() { .expect("Failed to create tokio runtime for telemetry"); let runtime_guard = runtime.enter(); - let (provider, _file_guard) = init_subscriber( + let (provider, stderr_guard, file_guard) = init_subscriber( &worker_config.logging, &worker_config.open_telemetry, &token, @@ -248,6 +250,8 @@ fn main() { { tracing::error!("Failed to flush OpenTelemetry provider: {err}"); } + drop(stderr_guard); + drop(file_guard); } fn raster_type_dispatch( diff --git a/geoengine/operators/src/source/gdal_worker_process/process_pool.rs b/geoengine/operators/src/source/gdal_worker_process/process_pool.rs index 27d00ded0..6a2d272da 100644 --- a/geoengine/operators/src/source/gdal_worker_process/process_pool.rs +++ b/geoengine/operators/src/source/gdal_worker_process/process_pool.rs @@ -461,6 +461,7 @@ impl GdalProcessPool { } } + #[allow(clippy::too_many_lines)] async fn broker_loop( mut rx: mpsc::Receiver, workers: Vec, @@ -526,11 +527,17 @@ impl GdalProcessPool { "Successfully respawned GDAL Worker {}", worker_id ); - let _ = b_tx.blocking_send(BrokerCommand::WorkerReplaced { - worker_id, - child_guard, - job_tx, - }); + if let Err(e) = + b_tx.blocking_send(BrokerCommand::WorkerReplaced { + worker_id, + child_guard, + job_tx, + }) + { + tracing::error!( + "Failed to send WorkerReplaced for GDAL Worker {worker_id}: {e:?}", + ); + } } Err(e) => { // Prevent silent failures if the respawn itself crashes @@ -571,6 +578,7 @@ impl GdalProcessPool { ); } } + tracing::error!("GDAL broker loop: channel closed, exiting"); } } diff --git a/geoengine/services/src/bin/geoengine-server.rs b/geoengine/services/src/bin/geoengine-server.rs index c1c624ec5..7bf222c29 100644 --- a/geoengine/services/src/bin/geoengine-server.rs +++ b/geoengine/services/src/bin/geoengine-server.rs @@ -38,14 +38,15 @@ pub async fn start_server() -> Result<()> { EnvFilter::try_new(&logging_config.log_spec).expect("to have a valid log spec"); // create a log layer for output to the console and add it to the registry - let registry = registry.with(console_layer_with_filter(console_filter)); + let (console_layer, console_guard) = console_layer_with_filter(console_filter); + let registry = registry.with(console_layer); // create a filter for the log message level in file output. Since the console_filter is not copy or clone, we have to create a new one. TODO: allow a different log level for file output. let file_filter = EnvFilter::try_new(&logging_config.log_spec).expect("to have a valid log spec"); // create a log layer for output to a file and add it to the registry - let (file_layer, _writer_drop_guard) = if logging_config.log_to_file { + let (file_layer, writer_drop_guard) = if logging_config.log_to_file { let (file_layer, writer_drop_guard) = file_layer_with_filter( &logging_config.filename_prefix, logging_config.log_directory.as_deref(), @@ -72,7 +73,14 @@ pub async fn start_server() -> Result<()> { // initialize the registry as the global tracing subscriber registry.init(); - geoengine_services::server::start_server(None).await + match geoengine_services::server::start_server(None).await { + Ok(()) => tracing::info!("Server stopped successfully"), + Err(err) => tracing::error!("Server stopped with error: {err}"), + } + drop(writer_drop_guard); + drop(console_guard); + + Ok(()) } fn open_telemetry_layer( @@ -111,18 +119,23 @@ where Ok(opentelemetry) } -fn console_layer_with_filter + 'static>(filter: F) -> impl Layer +fn console_layer_with_filter + 'static>( + filter: F, +) -> (impl Layer + use, WorkerGuard) where S: Subscriber, for<'a> S: LookupSpan<'a>, { - tracing_subscriber::fmt::layer() + // we use the non-blocking wrapper around stderr so that we don't block the worker on logging. The guard is dropped at the end of main to flush any remaining logs. + let (non_blocking_writer, guard) = tracing_appender::non_blocking(std::io::stderr()); + let layer = tracing_subscriber::fmt::layer() .pretty() .with_file(false) .with_target(true) .with_ansi(true) - .with_writer(std::io::stderr) - .with_filter(filter) + .with_writer(non_blocking_writer) + .with_filter(filter); + (layer, guard) } // we use a custom formatter because there are still format flags within spans even when `with_ansi` is false due to bug: https://github.com/tokio-rs/tracing/issues/1817 diff --git a/python/geoengine/types.py b/python/geoengine/types.py index d90159711..9700bbc8a 100644 --- a/python/geoengine/types.py +++ b/python/geoengine/types.py @@ -1749,8 +1749,14 @@ class GeoTransform: def __init__(self, x_min: float, y_max: float, x_pixel_size: float, y_pixel_size: float): """Initialize a new `GeoTransform`""" + # Note: We use the GeoTransform in API to address two types in the backend. + # The first type is the backend GeoTransform. In this case, the x_pixel_size is always positive + # and the y_pixel_size is always negative. + # The second type is the GdalGeoTransform. In this case, the x_pixel_size is always positive + # and the y_pixel_size is in most cases negative, but there are cases where it is positive. + # Therefore, we only check that x_pixel_size is positive and y_pixel_size is not zero. assert x_pixel_size > 0, "In Geo Engine, x_pixel_size is always positive." - assert y_pixel_size < 0, "In Geo Engine, y_pixel_size is always negative." + assert y_pixel_size != 0, "y_pixel_size must not be zero." self.x_min = x_min self.y_max = y_max