diff --git a/Cargo.lock b/Cargo.lock index 668797c..5d4eb18 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2102,7 +2102,7 @@ dependencies = [ [[package]] name = "opsqueue" -version = "0.36.0" +version = "0.37.0" dependencies = [ "anyhow", "arc-swap", @@ -2155,7 +2155,7 @@ dependencies = [ [[package]] name = "opsqueue_python" -version = "0.36.0" +version = "0.37.0" dependencies = [ "anyhow", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 7fac6ef..6105f84 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,7 +14,7 @@ hakari-package = "workspace-hack" [workspace.package] edition = "2024" -version = "0.36.0" +version = "0.37.0" [workspace.dependencies] anyhow = { version = "1.0.102" } diff --git a/libs/opsqueue_python/src/consumer.rs b/libs/opsqueue_python/src/consumer.rs index c6f8c59..b31dd9f 100644 --- a/libs/opsqueue_python/src/consumer.rs +++ b/libs/opsqueue_python/src/consumer.rs @@ -1,3 +1,4 @@ +use opsqueue::tracing::as_dyn_error; use std::future::IntoFuture; use std::sync::Arc; use std::time::Duration; @@ -199,13 +200,16 @@ impl ConsumerClient { } // In essence we 'catch `Exception` (but _not_ `BaseException` here) Err(CError(L(e))) => { - tracing::info!("Opsqueue consumer closing because of exception: {e:?}"); + tracing::info!( + error = as_dyn_error(&e), + "Opsqueue consumer closing because of exception" + ); return CError(L(e)); } Err(CError(R(err))) => { tracing::warn!( - "Opsqueue consumer encountered a Rust error, but will continue: {}", - err + error = as_dyn_error(&err), + "Opsqueue consumer encountered a Rust error, but will continue", ); } } diff --git a/libs/opsqueue_python/src/errors.rs b/libs/opsqueue_python/src/errors.rs index 76b4172..45f0f7d 100644 --- a/libs/opsqueue_python/src/errors.rs +++ b/libs/opsqueue_python/src/errors.rs @@ -70,7 +70,7 @@ pub type CPyResult = Result>; /// allowing things like `KeyboardInterrupt`, `SystemExit` or `MemoryError`, /// to trigger cleanup-and-exit. #[derive(thiserror::Error, Debug)] -#[error("Fatal Python exception: {0}")] +#[error("Fatal Python exception")] pub struct FatalPythonException(#[from] pub PyErr); impl From> for PyErr { diff --git a/opsqueue/app/main.rs b/opsqueue/app/main.rs index 0588974..5baac61 100644 --- a/opsqueue/app/main.rs +++ b/opsqueue/app/main.rs @@ -4,9 +4,9 @@ use opentelemetry_otlp::SpanExporter; use opentelemetry_resource_detectors::HostResourceDetector; use opentelemetry_resource_detectors::{OsResourceDetector, ProcessResourceDetector}; use opentelemetry_sdk::trace::{RandomIdGenerator, Sampler, SdkTracerProvider}; +use opsqueue::tracing::as_dyn_error; use opsqueue::{common::submission::db::periodically_cleanup_old, config::Config, prometheus}; use std::{ - error::Error, sync::{Arc, atomic::AtomicBool}, time::Duration, }; @@ -104,7 +104,7 @@ pub async fn async_main() { res = tokio::signal::ctrl_c() => { match res { Ok(()) => tracing::warn!("Received Ctrl-C signal"), - Err(ref err) => tracing::error!(error = err as &dyn Error, "Error while waiting for Ctrl-C signal"), + Err(ref err) => tracing::error!(error = as_dyn_error(err), "Error while waiting for Ctrl-C signal"), } }, }; @@ -138,6 +138,7 @@ fn init_sentry() -> sentry::ClientInitGuard { traces_sample_rate: 0.0, send_default_pii: true, release: sentry::release_name!(), + in_app_include: vec!["opsqueue::"], ..Default::default() }; sentry::init(options) diff --git a/opsqueue/src/common/errors.rs b/opsqueue/src/common/errors.rs index efc82db..6527810 100644 --- a/opsqueue/src/common/errors.rs +++ b/opsqueue/src/common/errors.rs @@ -16,11 +16,8 @@ use super::{ submission::{SubmissionCancelled, SubmissionCompleted, SubmissionFailed, SubmissionId}, }; -// #[derive(Error, Debug, Clone, Serialize, Deserialize)] -// #[error("Low-level database error: {0:?}")] -// pub struct DatabaseError(#[from] pub serde_error::Error); #[cfg_attr(feature = "server-logic", derive(Error, Debug))] -#[cfg_attr(feature = "server-logic", error("Low-level database error: {0:?}"))] +#[cfg_attr(feature = "server-logic", error("Low-level database error"))] #[cfg(feature = "server-logic")] pub struct DatabaseError(#[from] pub sqlx::Error); @@ -136,7 +133,7 @@ impl From> for E> { } #[derive(Error, Debug, PartialEq, Eq, Clone, Serialize, Deserialize)] -#[error("You are using Opsqueue incorrectly. Details: {0}")] +#[error("You are using Opsqueue incorrectly")] pub struct IncorrectUsage(#[from] pub E); #[derive(Error, Debug, PartialEq, Eq, Clone, Serialize, Deserialize)] diff --git a/opsqueue/src/common/mod.rs b/opsqueue/src/common/mod.rs index c685cd6..14fe5c5 100644 --- a/opsqueue/src/common/mod.rs +++ b/opsqueue/src/common/mod.rs @@ -51,18 +51,30 @@ impl std::fmt::Display for MaxSubmissions { } } +impl std::str::FromStr for MaxSubmissions { + type Err = ParseMaxSubmissionsError; + fn from_str(s: &str) -> Result { + let value: NonZero = s.parse()?; + Ok(MaxSubmissions::new(value)?) + } +} + #[derive(Debug, thiserror::Error)] pub enum ParseMaxSubmissionsError { #[error(transparent)] - NotANumber(#[from] std::num::ParseIntError), + NotANumber(std::num::ParseIntError), #[error(transparent)] - TooLarge(#[from] MaxSubmissionsTooLarge), + TooLarge(MaxSubmissionsTooLarge), } -impl std::str::FromStr for MaxSubmissions { - type Err = ParseMaxSubmissionsError; - fn from_str(s: &str) -> Result { - let value: NonZero = s.parse()?; - Ok(MaxSubmissions::new(value)?) +impl From for ParseMaxSubmissionsError { + fn from(value: std::num::ParseIntError) -> Self { + Self::NotANumber(value) + } +} + +impl From for ParseMaxSubmissionsError { + fn from(value: MaxSubmissionsTooLarge) -> Self { + Self::TooLarge(value) } } diff --git a/opsqueue/src/common/submission.rs b/opsqueue/src/common/submission.rs index 9cc469a..991744d 100644 --- a/opsqueue/src/common/submission.rs +++ b/opsqueue/src/common/submission.rs @@ -276,6 +276,7 @@ impl Submission { #[cfg(feature = "server-logic")] pub mod db { + use crate::tracing::as_dyn_error; use crate::{ common::{ MaxSubmissions, StrategicMetadataMap, @@ -1179,7 +1180,7 @@ pub mod db { } .await; if let Err(e) = res { - tracing::error!("Error during periodic cleanup: {}", e); + tracing::error!(error = as_dyn_error(&e), "Error during periodic cleanup"); } tokio::time::sleep(PERIODIC_CLEANUP_INTERVAL).await; } diff --git a/opsqueue/src/consumer/client.rs b/opsqueue/src/consumer/client.rs index 594f736..94abc81 100644 --- a/opsqueue/src/consumer/client.rs +++ b/opsqueue/src/consumer/client.rs @@ -1,6 +1,6 @@ +use crate::tracing::{anyhow_as_dyn_error, as_dyn_error}; use std::{ collections::HashMap, - error::Error, str::FromStr, sync::{ Arc, @@ -160,10 +160,15 @@ impl OuterClient { async fn initialize(&self) -> Client { tracing::info!("Initializing (or re-initializing) consumer client connection..."); (|| Client::new(&self.1)) - .retry(retry_policy()) - .notify(|err, duration| { tracing::debug!("Error establishing consumer client WS connection. (Will retry in {duration:?}). Details: {err:?}") }) - .await - .expect("Infinite retries should never return Err") + .retry(retry_policy()) + .notify(|err, duration| { + tracing::debug!( + error = anyhow_as_dyn_error(err), + "Error establishing consumer client WS connection. (Will retry in {duration:?})" + ); + }) + .await + .expect("Infinite retries should never return Err") } /// When `false` is returned, the next call to the client will attempt to restore the connection. @@ -356,7 +361,7 @@ impl Client { break; } Some(Err(e)) => { - tracing::error!("Opsqueue consumer client background task closing, reason: {e}"); + tracing::error!(error = as_dyn_error(&e), "Opsqueue consumer client background task closing, reason"); break; }, Some(Ok(msg)) => { @@ -389,7 +394,7 @@ impl Client { failure: "Requester disappeared".into(), }, ).await else { continue; }; - tracing::error!(error = &internal_error as &dyn Error, chunk_id = ?chunk_id, "Failed to return unhandled reservation to server"); + tracing::error!(error = as_dyn_error(&internal_error), chunk_id = ?chunk_id, "Failed to return unhandled reservation to server"); } }, } @@ -416,7 +421,7 @@ impl Client { failure: "Reservation expired".into(), }, ).await else { continue; }; - tracing::error!(error = &internal_error as &dyn Error, chunk_id = ?chunk_id, "Failed to return expired reservation to server"); + tracing::error!(error = as_dyn_error(&internal_error), chunk_id = ?chunk_id, "Failed to return expired reservation to server"); }, } } @@ -519,10 +524,10 @@ impl Client { #[derive(thiserror::Error, Debug)] pub enum InternalConsumerClientError { - #[error("Low-level error in the websocket connection: {0}")] + #[error("Low-level error in the websocket connection")] LowLevelWebsocketError(#[from] tokio_tungstenite::tungstenite::Error), #[error( - "The oneshot channel to receive a sync response to an earlier request was dropped before a response was received: {0}" + "The oneshot channel to receive a sync response to an earlier request was dropped before a response was received" )] OneshotSenderDropped(#[from] RecvError), #[error("Expected the sync response of kind {expected} but received {actual:?}")] diff --git a/opsqueue/src/consumer/server/conn.rs b/opsqueue/src/consumer/server/conn.rs index 42aa713..8b9cecd 100644 --- a/opsqueue/src/consumer/server/conn.rs +++ b/opsqueue/src/consumer/server/conn.rs @@ -1,3 +1,4 @@ +use crate::tracing::as_dyn_error; use std::{ sync::{ Arc, @@ -216,7 +217,7 @@ impl ConsumerConn { .await; match chunks_or_err { Err(e) => { - tracing::error!("Failed to reserve chunks: {}", e); + tracing::error!(error = as_dyn_error(&e), "Failed to reserve chunks"); match e { // In the unlikely event of a DatabaseError, // this is not the client's fault but _our_ fault. diff --git a/opsqueue/src/consumer/server/mod.rs b/opsqueue/src/consumer/server/mod.rs index cc0166c..5cd9a22 100644 --- a/opsqueue/src/consumer/server/mod.rs +++ b/opsqueue/src/consumer/server/mod.rs @@ -1,3 +1,4 @@ +use crate::tracing::{anyhow_as_dyn_error, as_dyn_error}; use std::{ collections::HashSet, sync::{Arc, Mutex}, @@ -124,12 +125,14 @@ async fn ws_accept_handler( Ok(()) => {} Err(e) if e.is_internal_error() => { tracing::error!( - "Closed websocket connection because of internal error, details: {e:?}" + error = as_dyn_error(&e), + "Closed websocket connection because of internal error" ); } Err(e) => { tracing::warn!( - "Closed websocket connection because of client error, details: {e:?}" + error = as_dyn_error(&e), + "Closed websocket connection because of client error" ); } } @@ -298,7 +301,10 @@ impl Completer { .await; match res { Ok(()) => {} - Err(err) => tracing::error!("Error in chunk completer: {err}"), + Err(err) => tracing::error!( + error = anyhow_as_dyn_error(&err), + "Error in chunk completer" + ), } } } diff --git a/opsqueue/src/db/mod.rs b/opsqueue/src/db/mod.rs index 835d683..4514b80 100644 --- a/opsqueue/src/db/mod.rs +++ b/opsqueue/src/db/mod.rs @@ -53,6 +53,7 @@ //! # Ok (()) } //! ``` +use crate::tracing::as_dyn_error; use std::{marker::PhantomData, num::NonZero, time::Duration}; use futures::future::BoxFuture; @@ -215,8 +216,11 @@ impl DBPools { }) .await .is_ok(), - Err(error) => { - tracing::error!("DB unhealthy; could not acquire DB connection: {error:?}"); + Err(err) => { + tracing::error!( + error = as_dyn_error(&err), + "DB unhealthy; could not acquire DB connection" + ); false } } diff --git a/opsqueue/src/object_store/mod.rs b/opsqueue/src/object_store/mod.rs index 12ee43e..8df2c4e 100644 --- a/opsqueue/src/object_store/mod.rs +++ b/opsqueue/src/object_store/mod.rs @@ -50,9 +50,10 @@ impl std::fmt::Display for ChunkType { #[derive(thiserror::Error, Debug)] pub enum ChunkRetrievalError { #[error( - "Failed to retrieve chunk ({submission_prefix}, {chunk_index}, {chunk_type}) from object store: {source}" + "Failed to retrieve chunk ({submission_prefix}, {chunk_index}, {chunk_type}) from object store" )] ObjectStoreError { + #[source] source: object_store::Error, submission_prefix: Box, chunk_index: chunk::ChunkIndex, @@ -63,19 +64,21 @@ pub enum ChunkRetrievalError { #[derive(thiserror::Error, Debug)] pub enum ChunkStorageError { #[error( - "Failed to store chunk ({submission_prefix}, {chunk_index}, {chunk_type}) to object store: {source}" + "Failed to store chunk ({submission_prefix}, {chunk_index}, {chunk_type}) to object store" )] ObjectStoreError { + #[source] source: object_store::Error, submission_prefix: Box, chunk_index: chunk::ChunkIndex, chunk_type: ChunkType, }, - #[error("Failed to read chunk element from stream/iterator at index {chunk_index}: ")] + #[error("Failed to read chunk element from stream/iterator at index {chunk_index}")] ChunkContentsEvalError { submission_prefix: Box, chunk_index: chunk::ChunkIndex, chunk_type: ChunkType, + #[source] source: anyhow::Error, }, } @@ -84,19 +87,20 @@ pub enum ChunkStorageError { pub enum ChunksStorageError { #[error(transparent)] ChunkStorageError(#[from] ChunkStorageError), - #[error("Failed to read chunk element from stream/iterator: {source}")] + #[error("Failed to read chunk element from stream/iterator")] ChunkContentsEvalError { submission_prefix: Box, chunk_type: ChunkType, + #[source] source: anyhow::Error, }, } #[derive(thiserror::Error, Debug)] pub enum NewObjectStoreClientError { - #[error("Failed to parse URL: {0}")] + #[error("Failed to parse URL")] UrlParseFailure(#[from] url::ParseError), - #[error("URL is not valid object store URL {0}")] + #[error("URL is not valid object store URL")] ObjectStoreUrlFailure(#[from] object_store::Error), } diff --git a/opsqueue/src/producer/client.rs b/opsqueue/src/producer/client.rs index 13de24a..4f1dc92 100644 --- a/opsqueue/src/producer/client.rs +++ b/opsqueue/src/producer/client.rs @@ -13,7 +13,7 @@ use crate::{ errors::{SubmissionNotCancellable, SubmissionNotFound, TooManyMatchingSubmissions}, submission::{SubmissionId, SubmissionStatus}, }, - tracing::CarrierMap, + tracing::{CarrierMap, as_dyn_error}, }; use super::common::InsertSubmission; @@ -92,7 +92,10 @@ impl Client { .retry(retry_policy()) .when(InternalProducerClientError::is_ephemeral) .notify(|err, dur| { - tracing::debug!("retrying error {err:?} with sleeping {dur:?}"); + tracing::debug!( + error = as_dyn_error(err), + "retrying error with sleeping {dur:?}" + ); }) .await } @@ -127,7 +130,10 @@ impl Client { .retry(retry_policy()) .when(InternalProducerClientError::is_ephemeral) .notify(|err, dur| { - tracing::debug!("retrying error {err:?} with sleeping {dur:?}"); + tracing::debug!( + error = as_dyn_error(err), + "retrying error with sleeping {dur:?}" + ); }) .await } @@ -189,7 +195,10 @@ impl Client { R(R(client_err)) => client_err.is_ephemeral(), }) .notify(|err, dur| { - tracing::debug!("retrying error {err:?} with sleeping {dur:?}"); + tracing::debug!( + error = as_dyn_error(err), + "retrying error with sleeping {dur:?}" + ); }) .await } @@ -221,7 +230,10 @@ impl Client { .retry(retry_policy()) .when(InternalProducerClientError::is_ephemeral) .notify(|err, dur| { - tracing::debug!("retrying error {err:?} with sleeping {dur:?}"); + tracing::debug!( + error = as_dyn_error(err), + "retrying error with sleeping {dur:?}" + ); }) .await } @@ -253,7 +265,10 @@ impl Client { .retry(retry_policy()) .when(InternalProducerClientError::is_ephemeral) .notify(|err, dur| { - tracing::debug!("retrying error {err:?} with sleeping {dur:?}"); + tracing::debug!( + error = as_dyn_error(err), + "retrying error with sleeping {dur:?}" + ); }) .await } @@ -307,7 +322,10 @@ impl Client { R(client_err) => client_err.is_ephemeral(), }) .notify(|err, dur| { - tracing::debug!("retrying error {err:?} with sleeping {dur:?}"); + tracing::debug!( + error = as_dyn_error(err), + "retrying error with sleeping {dur:?}" + ); }) .await } @@ -337,9 +355,9 @@ impl Client { #[derive(thiserror::Error, Debug)] pub enum InternalProducerClientError { - #[error("HTTP request failed: {0}")] + #[error("HTTP request failed")] HTTPClientError(#[from] reqwest::Error), - #[error("Error decoding JSON response: {0}")] + #[error("Error decoding JSON response")] ResponseDecodingError(#[from] serde_json::Error), #[error("Internal client received unexpected status: {0}")] UnexpectedStatus(StatusCode), diff --git a/opsqueue/src/producer/server.rs b/opsqueue/src/producer/server.rs index b0d76a1..74b8707 100644 --- a/opsqueue/src/producer/server.rs +++ b/opsqueue/src/producer/server.rs @@ -4,6 +4,7 @@ use crate::common::errors::E::{L, R}; use crate::common::submission::{self, SubmissionId}; use crate::common::{MaxSubmissions, StrategicMetadataMap}; use crate::db::{self, DBPools}; +use crate::tracing::anyhow_as_dyn_error; use axum::extract; use axum::extract::{Path, State}; use axum::http::StatusCode; @@ -90,7 +91,10 @@ struct ServerError(anyhow::Error); impl IntoResponse for ServerError { fn into_response(self) -> Response { - tracing::error!("Producer Server Error {:?}", self.0); + tracing::error!( + error = anyhow_as_dyn_error(&self.0), + "Producer Server Error" + ); ( StatusCode::INTERNAL_SERVER_ERROR, diff --git a/opsqueue/src/prometheus.rs b/opsqueue/src/prometheus.rs index 3f12719..8bc7f02 100644 --- a/opsqueue/src/prometheus.rs +++ b/opsqueue/src/prometheus.rs @@ -13,6 +13,7 @@ use axum_prometheus::{ use tokio_util::sync::CancellationToken; use crate::db::DBPools; +use crate::tracing::anyhow_as_dyn_error; pub const SUBMISSIONS_TOTAL_COUNTER: &str = "submissions_total_count"; pub const SUBMISSIONS_COMPLETED_COUNTER: &str = "submissions_completed_count"; @@ -233,7 +234,7 @@ pub async fn periodically_calculate_scaling_metrics( () = cancellation_token.cancelled() => break, () = tokio::time::sleep(METRICS_INTERVAL) => { if let Err(e) = calculate_scaling_metrics(db_pool).await { - tracing::error!("Error calculating scaling metrics: {}", e); + tracing::error!(error = anyhow_as_dyn_error(&e), "Error calculating scaling metrics"); } } } diff --git a/opsqueue/src/server.rs b/opsqueue/src/server.rs index e42be26..5f9ec42 100644 --- a/opsqueue/src/server.rs +++ b/opsqueue/src/server.rs @@ -1,4 +1,5 @@ //! Defines the HTTP endpoints that are used by both the `producer` and `consumer` APIs +use crate::tracing::as_dyn_error; use std::{ any::Any, mem, @@ -54,16 +55,18 @@ pub async fn serve_producer_and_consumer( Ok(addr) => { tracing::info!("Server listening on {addr}"); if let Some(pipe) = config.report_bound_port_pipe.take() - && let Err(err) = pipe.write_port(addr.port()) { - tracing::warn!( - "Failed to write bound port {} to pipe: {}", - addr.port(), - err - ); - } + && let Err(err) = pipe.write_port(addr.port()) + { + tracing::warn!( + error = as_dyn_error(&err), + "Failed to write bound port {} to pipe", + addr.port(), + ); + } } Err(err) => tracing::warn!( - "Could not get locally bound address of the server, tried binding on {server_addr}: {err}" + error = as_dyn_error(&err), + "Could not get locally bound address of the server, tried binding on {server_addr}" ), } axum::serve(listener, router) @@ -72,8 +75,14 @@ pub async fn serve_producer_and_consumer( Ok(()) }) .retry(retry_policy()) - .notify(|e, d| tracing::error!("Error when binding server address: {e:?}, retrying in {d:?}")) - .await.inspect_err(|_|{ + .notify(|e, d| { + tracing::error!( + error = as_dyn_error(e), + "Error when binding server address, retrying in {d:?}" + ); + }) + .await + .inspect_err(|_| { // Drop the pipe after the server start retries have been exhausted. This ensures that the // parent process can safely block on reading from the pipe. mem::drop(config.report_bound_port_pipe.take()); diff --git a/opsqueue/src/tracing.rs b/opsqueue/src/tracing.rs index 0ba300a..0a6c5fb 100644 --- a/opsqueue/src/tracing.rs +++ b/opsqueue/src/tracing.rs @@ -4,6 +4,7 @@ use opentelemetry::{Context, propagation::TextMapCompositePropagator}; use opentelemetry_http::{HeaderExtractor, HeaderInjector}; use opentelemetry_sdk::propagation::{BaggagePropagator, TraceContextPropagator}; use rustc_hash::FxHashMap; +use std::error::Error; #[must_use] pub fn context_from_headers(headers: &http::HeaderMap) -> Context { @@ -57,3 +58,31 @@ pub fn propagator() -> TextMapCompositePropagator { Box::new(TraceContextPropagator::new()), ]) } + +/// Convenient function for converting an error into `dyn Error` for tracing. +/// +/// When logging an error, convert the error to `dyn Error` using this function +/// and assign it to the `error` field, rather than displaying the error, so +/// that the source chain of the error is preserved. +/// +/// ```ignore +/// // Good: +/// tracing::error!(error = as_dyn_error(e), "operation failed"); +/// +/// // Bad: +/// tracing::error!("operation failed: {e}"); +/// tracing::error!(error = %e, "operation failed"); +/// ``` +pub fn as_dyn_error(err: &T) -> &(dyn 'static + Error) +where + T: 'static + Error, +{ + err +} + +/// [`anyhow::Error`] does not implement [`Error`], so this function is +/// separate from [`as_dyn_error`]. +#[must_use] +pub fn anyhow_as_dyn_error(err: &anyhow::Error) -> &(dyn 'static + Error) { + err.as_ref() +}