diff --git a/crates/trident/src/logging/operation_context.rs b/crates/trident/src/logging/operation_context.rs index fb9451b3ae..e7689df6d4 100644 --- a/crates/trident/src/logging/operation_context.rs +++ b/crates/trident/src/logging/operation_context.rs @@ -43,6 +43,19 @@ use trident_api::{config::Operations, error::TridentError}; use crate::command_kind::CommandKind; +/// Message used by `server::tridentserver::TridentServer::try_acquire_read_lock`/ +/// `try_acquire_write_lock` for the `Status::unavailable` returned when +/// connection-lock contention blocks a request. `grpc_client::is_transport_failure` +/// matches on this exact message to recognize this as a deliberate +/// admission-control rejection -- the daemon DID receive and answer the +/// request -- rather than a genuine transport-level failure, even though +/// tonic uses the same `Code::Unavailable` for both. +pub(crate) const CONNECTION_LOCK_BUSY_MESSAGE: &str = "Trident is busy"; + +/// Same as [`CONNECTION_LOCK_BUSY_MESSAGE`], but for the servicing-lock +/// contention rejections in `servicing_request`/`reading_request`. +pub(crate) const SERVICING_LOCK_BUSY_MESSAGE: &str = "Servicing is active"; + /// Maps a base command name plus its requested `Operations` to the same /// naming convention gRPC's `servicing_request` already uses for /// stage/finalize granularity (e.g. `"install"` vs `"install_stage"` vs diff --git a/crates/trident/src/logging/tracestream.rs b/crates/trident/src/logging/tracestream.rs index 21f9a1075b..bea7179cbd 100644 --- a/crates/trident/src/logging/tracestream.rs +++ b/crates/trident/src/logging/tracestream.rs @@ -241,14 +241,26 @@ impl TraceStream { /// servicing ID yet), so it's still worth attaching even when /// `skip_installation_id` is true. /// - /// Safe to call from anywhere, any number of times, before any point - /// that wants either ID attached: this is the single implementation - /// shared by every read-only-in-the-common-case attach call site (the - /// CLI's dispatch, the daemon's startup attach, and the daemon's - /// per-request backstop), so a correctness fix to this logic only - /// needs to happen once. + /// Refresh servicing IDs only before servicing starts or while holding + /// its lock. Rejected requests must use the installation-only path so + /// a stale datastore read cannot overwrite an active operation's ID. /// Known edge case: multiboot finalize may prewarm the source `servicing_id`. pub fn attach_ids_if_present(&self, datastore_path: &Path, skip_installation_id: bool) { + self.attach_ids_from_path(datastore_path, skip_installation_id, true); + } + + /// Attaches or migrates the installation ID without touching the shared + /// servicing ID. Safe for rejected requests outside the servicing lock. + pub fn attach_installation_id_if_present(&self, datastore_path: &Path) { + self.attach_ids_from_path(datastore_path, false, false); + } + + fn attach_ids_from_path( + &self, + datastore_path: &Path, + skip_installation_id: bool, + attach_servicing_id: bool, + ) { if !datastore_path.exists() { return; } @@ -275,7 +287,9 @@ impl TraceStream { } } - self.attach_servicing_id_if_present(&datastore); + if attach_servicing_id { + self.attach_servicing_id_if_present(&datastore); + } } /// Ensures `datastore` has an installation ID -- creating one if this @@ -987,6 +1001,34 @@ mod tests { ); } + #[test] + fn test_installation_only_attachment_preserves_active_servicing_id() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("datastore.sqlite"); + let mut datastore = DataStore::open_or_create(&path).unwrap(); + let installation = datastore.ensure_installation_id().unwrap().to_string(); + let persisted_servicing = datastore.ensure_servicing_id().unwrap().to_string(); + drop(datastore); + + let stream = TraceStream::default(); + stream.set_servicing_id("active-stage".to_string()); + stream.attach_installation_id_if_present(&path); + assert_eq!( + *stream.installation_id_handle().read().unwrap(), + Some(installation) + ); + assert_eq!( + *stream.servicing_id_handle().read().unwrap(), + Some("active-stage".to_string()) + ); + + stream.attach_ids_if_present(&path, false); + assert_eq!( + *stream.servicing_id_handle().read().unwrap(), + Some(persisted_servicing) + ); + } + #[test] fn test_tracestream() { let temp_dir = tempfile::tempdir().unwrap(); diff --git a/crates/trident/src/server/tridentserver/mod.rs b/crates/trident/src/server/tridentserver/mod.rs index 60b705e602..fdb679c84f 100644 --- a/crates/trident/src/server/tridentserver/mod.rs +++ b/crates/trident/src/server/tridentserver/mod.rs @@ -160,21 +160,33 @@ impl TridentServer { /// Tries to acquire a read lock on the server's RwLock. If the lock /// cannot be acquired, returns a gRPC Status indicating that the server is /// busy. + /// + /// Intentionally fires no telemetry (`command_start`/`command_error`) for + /// this rejection, unlike `reject_invalid_argument`/`reject_invalid_field`: + /// a connection-lock contention failure isn't a + /// distinct servicing outcome the way a malformed request is -- it's + /// pure admission control, happens before `refresh_ids` would + /// even run, and (unlike a bad request) the caller is expected to retry + /// the exact same request rather than fix anything, so a low-value, + /// high-volume `command_error` stream isn't worth adding here. #[cfg(feature = "grpc-preview")] fn try_acquire_read_lock(&self) -> Result, Status> { self.rwlock.clone().try_read_owned().map_err(|_| { warn!("Trident is busy, cannot acquire read connection lock"); - Status::unavailable("Trident is busy") + Status::unavailable(operation_context::CONNECTION_LOCK_BUSY_MESSAGE) }) } /// Tries to acquire a write lock on the server's RwLock. If the lock /// cannot be acquired, returns a gRPC Status indicating that the server is /// busy. + /// + /// See the telemetry note on [`Self::try_acquire_read_lock`]: this + /// rejection is intentionally untelemetered for the same reason. fn try_acquire_write_lock(&self) -> Result, Status> { self.rwlock.clone().try_write_owned().map_err(|_| { warn!("Trident is busy, cannot acquire write connection lock"); - Status::unavailable("Trident is busy") + Status::unavailable(operation_context::CONNECTION_LOCK_BUSY_MESSAGE) }) } @@ -259,14 +271,14 @@ impl TridentServer { // Re-check for a persisted installation ID and current servicing // ID before this request fires its own command_start (below, via - // run_with_operation). server_main's daemon-startup attach only + // run_command). server_main's daemon-startup attach only // ever runs once, at startup -- so a request that arrives before // any datastore exists (e.g. this daemon's very first install) // would otherwise never see one, even after that request's own // handler goes on to create the datastore. // // Deferred into the `f` closure below (called right before - // run_with_operation, preserving "before command_start") rather + // run_command, preserving "before command_start") rather // than run synchronously here: `attach_ids_if_present` opens the // datastore, and datastore connections use a 5-second busy // timeout, so lock contention could otherwise stall this @@ -297,29 +309,33 @@ impl TridentServer { let kind_for_closure = kind.clone(); let f = move || { Self::refresh_ids(&tracestream_for_ids, &datastore_path_for_ids); - operation_context::run_with_operation( - &kind_for_closure, - operation_context::OperationSource::Daemon, - || { - let result = f(); - if let Ok((ExitKind::NeedsReboot, ..)) = &result { - operation_context::save_reboot_operation(); - } - result - }, - ) + run_servicing_command(&kind_for_closure, || { + let result = f(); + if let Ok((ExitKind::NeedsReboot, ..)) = &result { + operation_context::save_reboot_operation(); + } + result + }) }; // Create the gRPC response channel let (tx, rx) = mpsc::unbounded_channel(); - // Try to acquire the servicing lock + // Try to acquire the servicing lock. Rejected here, after + // `refresh_ids` above but before the `run_command` + // closure `f` (built above) ever runs, this is intentionally + // untelemetered for the same reason as the connection-lock + // rejections in `try_acquire_read_lock`/`try_acquire_write_lock`: + // it's admission control, not a distinct servicing outcome, and + // the caller is expected to retry rather than fix anything. let Some(servicing_guard) = self.servicing_manager.try_lock_servicing() else { warn!( "Request '{}' blocked because servicing is active", kind.as_str() ); - return Err(Status::unavailable("Servicing is active")); + return Err(Status::unavailable( + operation_context::SERVICING_LOCK_BUSY_MESSAGE, + )); }; // Set up log forwarding. Logs will be sent over the gRPC channel. @@ -423,15 +439,31 @@ impl TridentServer { // request. let _guard = self.try_acquire_read_lock()?; - // Try to acquire the servicing read lock + // Try to acquire the servicing read lock. Same intentional + // telemetry gap as the connection-lock rejections in + // `try_acquire_read_lock`/`try_acquire_write_lock` and the + // servicing-lock rejection in `servicing_request`: it's admission + // control rather than a distinct read outcome, and the caller is + // expected to retry rather than fix anything. let Some(servicing_guard) = self.servicing_manager.try_lock_reading() else { warn!( "Read request '{}' blocked because servicing is active", name ); - return Err(Status::unavailable("Servicing is active")); + return Err(Status::unavailable( + operation_context::SERVICING_LOCK_BUSY_MESSAGE, + )); }; + // Read requests (e.g. `get_servicing_state`, `check_rollback`) are + // intentionally left untelemetered -- like their CLI counterparts + // (`get`, `validate`, `diagnose`, etc.; see `run_trident` in + // main.rs), none of them emit `command_start`/`command_error` or + // any other `metric_name` event, so there's no need to prep the + // TraceStream's installation ID/database ID (`refresh_ids`, used + // by `servicing_request` for exactly that reason) or wrap `f` in + // `operation_context::run_command` -- it just runs directly here. + // Execute the reading function Ok(Response::new( ServicingManager::spawn_reading_task(servicing_guard, f) @@ -444,6 +476,15 @@ impl TridentServer { } } +fn run_servicing_command( + kind: &CommandKind, + f: impl FnOnce() -> Result, +) -> Result { + operation_context::run_command_if(kind, operation_context::OperationSource::Daemon, f, |_| { + kind.is_servicing() + }) +} + #[cfg(feature = "grpc-preview")] fn trident_error_to_status(err: TridentError) -> Status { let code = match err.kind() { @@ -458,3 +499,72 @@ fn trident_error_to_status(err: TridentError) -> Status { Status::new(code, format!("{err:?}")) } + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tracing::{field::Visit, Event, Subscriber}; + use tracing_subscriber::{layer::Context, prelude::*, Layer, Registry}; + + #[derive(Clone, Default)] + struct CommandErrors(Arc); + + impl Layer for CommandErrors { + fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { + struct Metric(bool); + impl Visit for Metric { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + if field.name() == "metric_name" && value == "command_error" { + self.0 = true; + } + } + + fn record_debug(&mut self, _: &tracing::field::Field, _: &dyn std::fmt::Debug) {} + } + let mut metric = Metric(false); + event.record(&mut metric); + if metric.0 { + self.0.fetch_add(1, Ordering::SeqCst); + } + } + } + + #[test] + fn test_request_errors_follow_servicing_classification() { + for kind in [CommandKind::check_root(), CommandKind::commit()] { + let errors = CommandErrors::default(); + let subscriber = Registry::default().with(errors.clone()); + let result: Result<(), TridentError> = + tracing::subscriber::with_default(subscriber, || { + run_servicing_command(&kind, || { + let (_, active, source) = operation_context::current().unwrap(); + assert_eq!(active.as_str(), kind.as_str()); + assert_eq!(source, operation_context::OperationSource::Daemon); + Err(TridentError::internal("request failed")) + }) + }); + assert!(result.is_err()); + assert_eq!( + errors.0.load(Ordering::SeqCst), + usize::from(kind.is_servicing()) + ); + assert!(operation_context::current().is_none()); + } + } + + #[test] + fn test_nonservicing_request_panic_is_still_reported() { + let errors = CommandErrors::default(); + let subscriber = Registry::default().with(errors.clone()); + let result = tracing::subscriber::with_default(subscriber, || { + panic::catch_unwind(|| { + let _: Result<(), TridentError> = + run_servicing_command(&CommandKind::check_root(), || panic!("request panic")); + }) + }); + assert!(result.is_err()); + assert_eq!(errors.0.load(Ordering::SeqCst), 1); + assert!(operation_context::current().is_none()); + } +} diff --git a/crates/trident/src/server/tridentserver/services/install.rs b/crates/trident/src/server/tridentserver/services/install.rs index 25fca497fd..22f15411c3 100644 --- a/crates/trident/src/server/tridentserver/services/install.rs +++ b/crates/trident/src/server/tridentserver/services/install.rs @@ -27,17 +27,27 @@ impl InstallService for TridentServer { ) -> Result, Status> { let req = request.into_inner(); let Some(staging) = req.stage else { - return Err(Status::invalid_argument("Missing staging configuration")); + return Err(self.reject_invalid_argument( + &CommandKind::install(), + "stage", + "Missing staging configuration", + )); }; let Some(host_config) = staging.config else { - return Err(Status::invalid_argument( + return Err(self.reject_invalid_argument( + &CommandKind::install(), + "stage.config", "Missing host configuration in staging configuration", )); }; let Some(finalize) = req.finalize else { - return Err(Status::invalid_argument("Missing finalize configuration")); + return Err(self.reject_invalid_argument( + &CommandKind::install(), + "finalize", + "Missing finalize configuration", + )); }; let data_store_path = self.agent_config.datastore_path().to_owned(); @@ -74,7 +84,9 @@ impl InstallService for TridentServer { let req = request.into_inner(); let Some(host_config) = req.config else { - return Err(Status::invalid_argument( + return Err(self.reject_invalid_argument( + &CommandKind::install_stage(), + "config", "Missing host configuration in staging configuration", )); }; diff --git a/crates/trident/src/server/tridentserver/services/mod.rs b/crates/trident/src/server/tridentserver/services/mod.rs index e58bd12c6c..2fbe391e11 100644 --- a/crates/trident/src/server/tridentserver/services/mod.rs +++ b/crates/trident/src/server/tridentserver/services/mod.rs @@ -1,6 +1,12 @@ +use tonic::Status; +use trident_api::error::{InvalidInputError, TridentError}; use trident_proto::v1::{RebootHandling, RebootManagement}; -use crate::server::tridentserver::RebootDecision; +use crate::{ + command_kind::CommandKind, + logging::operation_context, + server::tridentserver::{RebootDecision, TridentServer}, +}; mod commit; mod rollback; @@ -17,6 +23,93 @@ mod status; #[cfg(feature = "grpc-preview")] mod validation; +impl TridentServer { + /// Rejects a gRPC request whose payload failed pre-dispatch validation + /// (missing field, unparsable Host Configuration, etc.) before + /// `servicing_request` ever runs. Without this, a rejected request left + /// no telemetry trace at all -- no `command_start`/`command_error` -- + /// unlike every request that makes it far enough to be serviced. Fires + /// both metrics for a synthetic, immediately-failed operation tagged with + /// the same `command` name the real dispatch would have used, then + /// returns the `Status` to send back to the caller. + /// + /// Attaches the installation ID when available, but does not refresh + /// the shared servicing ID: rejection runs outside the servicing lock + /// and must not overwrite the active operation's identity. + /// + /// `#[track_caller]` so `TridentError::new` below (itself + /// `#[track_caller]`) attributes this error's `location` to whichever + /// service handler actually rejected the request, not to this shared + /// helper's own line -- otherwise every rejection from every RPC would + /// report the identical, uninformative `location`. + #[track_caller] + fn reject_invalid_argument( + &self, + command: &CommandKind, + field: &str, + message: impl Into, + ) -> Status { + let error = TridentError::new(InvalidInputError::MissingRequestField { + field: field.to_owned(), + }); + // Unlike `servicing_request`'s closures, this runs directly on the + // async gRPC handler's Tokio worker thread, not inside + // `spawn_blocking` -- but `run_command` still synchronously fires + // tracing events, and a configured remote telemetry sender does a + // blocking `reqwest::blocking` POST from inside that same call + // (`TraceSender::on_event`). `block_in_place` tells the multi-threaded + // runtime this thread is about to block, so it can hand off other + // queued work to another worker instead of stalling behind it -- a + // burst of malformed requests can no longer starve the runtime. + // Installation-ID attachment runs inside this same closure (rather than + // before it) because it does its own synchronous SQLite I/O -- + // including up to a 5-second busy-timeout wait -- so it needs the + // same "about to block" signal to the runtime as `run_command`. + let _ = tokio::task::block_in_place(|| { + self.tracestream + .attach_installation_id_if_present(self.agent_config.datastore_path()); + operation_context::run_command( + command, + operation_context::OperationSource::Daemon, + || Err::<(), _>(error), + ) + }); + Status::invalid_argument(message.into()) + } + + /// Same as [`Self::reject_invalid_argument`], but for a field that is + /// present yet fails to parse or otherwise doesn't satisfy the + /// request's requirements (e.g. `stream_disk`'s image URL) rather than + /// a missing field. + /// + /// `#[track_caller]` for the same reason as + /// [`Self::reject_invalid_argument`]. + #[track_caller] + fn reject_invalid_field( + &self, + command: &CommandKind, + field: &str, + reason: impl Into, + message: impl Into, + ) -> Status { + let error = TridentError::new(InvalidInputError::InvalidRequestField { + field: field.to_owned(), + reason: reason.into(), + }); + // See the `block_in_place` comment in `reject_invalid_argument`. + let _ = tokio::task::block_in_place(|| { + self.tracestream + .attach_installation_id_if_present(self.agent_config.datastore_path()); + operation_context::run_command( + command, + operation_context::OperationSource::Daemon, + || Err::<(), _>(error), + ) + }); + Status::invalid_argument(message.into()) + } +} + /// Returns a `RebootDecision` indicating whether Trident can perform a reboot /// given a provided optional RebootManagement struct. fn reboot_allowed(reboot_opt: &Option) -> RebootDecision { diff --git a/crates/trident/src/server/tridentserver/services/rollback.rs b/crates/trident/src/server/tridentserver/services/rollback.rs index 61cf030451..8baba9464b 100644 --- a/crates/trident/src/server/tridentserver/services/rollback.rs +++ b/crates/trident/src/server/tridentserver/services/rollback.rs @@ -61,10 +61,18 @@ impl RollbackService for TridentServer { ) -> Result, Status> { let req = request.into_inner(); let Some(stage) = req.stage else { - return Err(Status::invalid_argument("Missing stage configuration")); + return Err(self.reject_invalid_argument( + &CommandKind::rollback(), + "stage", + "Missing stage configuration", + )); }; let Some(finalize) = req.finalize else { - return Err(Status::invalid_argument("Missing finalize configuration")); + return Err(self.reject_invalid_argument( + &CommandKind::rollback(), + "finalize", + "Missing finalize configuration", + )); }; let data_store_path = self.agent_config.datastore_path().to_owned(); diff --git a/crates/trident/src/server/tridentserver/services/streaming.rs b/crates/trident/src/server/tridentserver/services/streaming.rs index 6e36a94a0e..e8a4a70b19 100644 --- a/crates/trident/src/server/tridentserver/services/streaming.rs +++ b/crates/trident/src/server/tridentserver/services/streaming.rs @@ -20,9 +20,17 @@ impl StreamingService for TridentServer { let req = request.into_inner(); // Parse the image URL from the request, returning an error if it is invalid. - let url = Url::parse(&req.image_url).map_err(|e| { - Status::invalid_argument(format!("Invalid image URL '{}': {}", req.image_url, e)) - })?; + let url = match Url::parse(&req.image_url) { + Ok(url) => url, + Err(e) => { + return Err(self.reject_invalid_field( + &CommandKind::stream_disk(), + "image_url", + e.to_string(), + format!("Invalid image URL '{}': {}", req.image_url, e), + )); + } + }; // If the image hash is not provided, we use the constant for ignored checksum. let image_hash = req diff --git a/crates/trident/src/server/tridentserver/services/update.rs b/crates/trident/src/server/tridentserver/services/update.rs index 7f2edb9ab9..4f8e25f076 100644 --- a/crates/trident/src/server/tridentserver/services/update.rs +++ b/crates/trident/src/server/tridentserver/services/update.rs @@ -26,17 +26,27 @@ impl UpdateService for TridentServer { ) -> Result, Status> { let req = request.into_inner(); let Some(staging) = req.stage else { - return Err(Status::invalid_argument("Missing staging configuration")); + return Err(self.reject_invalid_argument( + &CommandKind::update(), + "stage", + "Missing staging configuration", + )); }; let Some(host_config) = staging.config else { - return Err(Status::invalid_argument( + return Err(self.reject_invalid_argument( + &CommandKind::update(), + "stage.config", "Missing host configuration in staging configuration", )); }; let Some(finalize) = req.finalize else { - return Err(Status::invalid_argument("Missing finalize configuration")); + return Err(self.reject_invalid_argument( + &CommandKind::update(), + "finalize", + "Missing finalize configuration", + )); }; let data_store_path = self.agent_config.datastore_path().to_owned(); @@ -73,7 +83,9 @@ impl UpdateService for TridentServer { let req = request.into_inner(); let Some(host_config) = req.config else { - return Err(Status::invalid_argument( + return Err(self.reject_invalid_argument( + &CommandKind::update_stage(), + "config", "Missing host configuration in staging configuration", )); }; diff --git a/crates/trident/src/server/tridentserver/services/validation.rs b/crates/trident/src/server/tridentserver/services/validation.rs index 0d31396507..71954d2eb8 100644 --- a/crates/trident/src/server/tridentserver/services/validation.rs +++ b/crates/trident/src/server/tridentserver/services/validation.rs @@ -24,6 +24,13 @@ impl ValidationService for TridentServer { // whenever without doing any lock checks. info!("Received Host Configuration validation request"); let Some(host_config) = request.into_inner().config else { + // Unlike `install`/`update`/`rollback`'s `reject_invalid_argument` + // calls, this deliberately constructs the `Status` directly + // instead: `validate_host_configuration` is a read-only RPC (see + // the comment above) with no `command_start`/`command_error` + // telemetry of its own, and routing this rejection through + // `reject_invalid_argument` would give it exactly that -- + // inconsistent with every other outcome of this RPC. return Err(Status::invalid_argument( "Missing host configuration in staging configuration", )); @@ -32,6 +39,7 @@ impl ValidationService for TridentServer { let error = validation::validate_host_config_string(&host_config.config) .err() .map(ProtoTridentError::from); + Ok(Response::new(ValidateHostConfigurationResponse { ok: error.is_none(), error, diff --git a/crates/trident_api/src/error.rs b/crates/trident_api/src/error.rs index 8270313b93..3442f1a7a8 100644 --- a/crates/trident_api/src/error.rs +++ b/crates/trident_api/src/error.rs @@ -217,6 +217,12 @@ pub enum InvalidInputError { #[error("Cannot find history file")] HistoryFileNotFound, + #[error("Missing required field '{field}' in request")] + MissingRequestField { field: String }, + + #[error("Invalid value for field '{field}' in request: {reason}")] + InvalidRequestField { field: String, reason: String }, + #[error("Cannot update host since it is not provisioned")] HostNotProvisioned,