diff --git a/Cargo.toml b/Cargo.toml index d83f88e0c..13fedc140 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -54,6 +54,7 @@ nix = { version = "0.30.1", features = [ "user", "socket", "signal", + "time", ], default-features = false } oci-client = "0.15.0" once_cell = "1.19" diff --git a/crates/trident/src/engine/manual_rollback/mod.rs b/crates/trident/src/engine/manual_rollback/mod.rs index 092267621..e22156de0 100644 --- a/crates/trident/src/engine/manual_rollback/mod.rs +++ b/crates/trident/src/engine/manual_rollback/mod.rs @@ -96,6 +96,27 @@ pub fn execute_rollback( requested_rollback_kind: ManualRollbackRequestKind, allowed_operations: &Operations, ) -> Result<(ExitKind, ServicingType), TridentError> { + // Fired once per *logical* rollback, not once per call: a + // finalize-only call (the second step of the CLI/gRPC two-step + // stage-then-finalize flow) resumes a rollback that was already + // staged -- and therefore already reported -- by an earlier + // stage-having call (stage-only or combined), so it isn't a new + // start. Gating on `has_stage()` fires exactly for the calls that + // genuinely begin a rollback (stage-only and combined); a + // finalize-only call fires nothing here. Whatever identifying context + // is known this early is included -- the specific A/B-vs-runtime + // `ManualRollbackKind` isn't determined until the stage/finalize + // logic below runs, so it isn't included here. + if allowed_operations.has_stage() { + tracing::info!( + metric_name = "manual_rollback_start", + requested_rollback_kind = format!("{:?}", requested_rollback_kind), + servicing_state = format!("{:?}", datastore.host_status().servicing_state), + stage = allowed_operations.has_stage(), + finalize = allowed_operations.has_finalize(), + ); + } + // Tracks the rollback kind actually staged this call, so the trailing // "stage completed, finalize not requested this call" return below can // report it instead of a generic NoActiveServicing. Stays None when @@ -294,6 +315,25 @@ fn finalize_rollback( host_status.spec_old = Default::default(); host_status.servicing_state = ServicingState::Provisioned; })?; + + // Unlike the A/B rollback case below, a runtime rollback requires no + // reboot, so this never reaches `engine::rollback`'s post-reboot + // boot-validation flow -- the only place `manual_rollback_success` + // is otherwise fired (and only for the `ManualRollbackAbFinalized` + // state). Without this, a runtime rollback would emit + // `manual_rollback_start` but no matching success signal at all. + info!("Manual rollback of runtime update succeeded"); + tracing::info!( + metric_name = "manual_rollback_runtime_success", + value = true + ); + + // Persistence happens in the caller (execute_rollback), after this + // function returns -- not here. This function's own outcome metric + // has already fired above by the time that happens, so the + // archived metrics file still includes it; persisting again here + // as well would just archive the same (or, with second-resolution + // filenames, a second) copy. return Ok(rollback_exit_kind); } diff --git a/crates/trident/src/engine/manual_rollback/utils.rs b/crates/trident/src/engine/manual_rollback/utils.rs index 04df1db6b..a22a3d2af 100644 --- a/crates/trident/src/engine/manual_rollback/utils.rs +++ b/crates/trident/src/engine/manual_rollback/utils.rs @@ -23,6 +23,7 @@ lazy_static! { } /// ManualRollbackRequestKind represents the kind of manual rollback request. +#[derive(Debug, Clone, Copy)] pub enum ManualRollbackRequestKind { RollbackOnlyIfNextIsRuntimeUpdate, RollbackAvailableAbUpdate, diff --git a/crates/trident/src/engine/runtime_update.rs b/crates/trident/src/engine/runtime_update.rs index 95cc4da4e..dc70b5445 100644 --- a/crates/trident/src/engine/runtime_update.rs +++ b/crates/trident/src/engine/runtime_update.rs @@ -11,7 +11,9 @@ use trident_api::{ use crate::{ datastore::DataStore, engine::{self, EngineContext, EngineContextParams}, - health, monitor_metrics, ExitKind, + health, + logging::operation_context, + monitor_metrics, ExitKind, }; use super::Subsystem; @@ -91,10 +93,51 @@ pub(crate) fn finalize_update( if let Err(e) = finalize_result { error!("Runtime update finalize failed with message:\n{e:?}"); // Attempt an auto-rollback - return rollback(subsystems, state, update_start_time).message(format!( + let rollback_result = rollback(subsystems, state, update_start_time).message(format!( "Auto-rollback was triggered by runtime update failure:\n{e:?}" )); + // Persist here (not inside finalize_or_rollback_runtime_update -- + // see the comment on that function) now that the auto-rollback's + // own outcome is known, so the archived metrics file actually + // includes it either way -- including a *failed* auto-rollback. + // Explicitly fire `command_error` for that final outcome *before* + // persisting (rather than leaving it to `run_command`/ + // `run_command_if`, further up the call stack, which would only + // see it well after this archive is already written): this is a + // no-op if that outer wrapper has (unusually) already reported an + // error for this operation, and it's the same call that wrapper + // would otherwise make on its own once `rollback_result` + // eventually reaches it as an `Err`, so this doesn't introduce a + // second, duplicate `command_error` event. + if let Err(ref outcome_error) = rollback_result { + operation_context::report_command_error(outcome_error); + } + engine::persist_background_log_and_metrics( + &state.host_status().spec.trident.datastore_path, + None, + state.host_status().servicing_state, + ); + return rollback_result; } + + // Unlike A/B update and clean install, a runtime update requires no + // reboot, so success can be confirmed synchronously right here instead + // of via the post-reboot boot-validation flow in `engine::rollback` + // (which only ever sees `CleanInstallFinalized`/`AbUpdateFinalized`/ + // `ManualRollbackAbFinalized` -- runtime update/rollback finalize and + // return to `Provisioned` without ever going through that flow). + info!("Runtime update succeeded"); + tracing::info!(metric_name = "runtime_update_success", value = true); + + // Persist *after* the success metric above so the archived metrics + // file on the target OS actually includes this event, not just a + // snapshot taken before it was ever emitted. + engine::persist_background_log_and_metrics( + &state.host_status().spec.trident.datastore_path, + None, + state.host_status().servicing_state, + ); + finalize_result } @@ -211,12 +254,13 @@ fn finalize_or_rollback_runtime_update( ); } - // Persist the Trident background log and metrics file to the updated target OS - engine::persist_background_log_and_metrics( - &state.host_status().spec.trident.datastore_path, - None, - state.host_status().servicing_state, - ); + // Persistence moved to each caller (finalize_update, and + // manual_rollback::finalize_rollback for the manual-rollback case), + // right after their own final outcome metric fires -- this function + // returning `Ok` here does not by itself mean any of those metrics + // have been emitted yet, so persisting here could snapshot the + // metrics file before its own caller's success/rollback event was + // ever appended to it. Ok(ExitKind::Done) } diff --git a/crates/trident/src/grpc_client/mod.rs b/crates/trident/src/grpc_client/mod.rs index c67a520a9..781b1eaa2 100644 --- a/crates/trident/src/grpc_client/mod.rs +++ b/crates/trident/src/grpc_client/mod.rs @@ -1,20 +1,24 @@ -use std::process::ExitCode; +use std::{cell::Cell, process::ExitCode}; use anyhow::{bail, Context, Error}; use log::error; use tokio::fs; use tokio::runtime::Builder; +use tonic::Code; + +use trident_api::error::{InternalError, TridentError}; use crate::{ - cli::{ClientArgs, ClientCommands, TridentExitCodes}, - ExitKind, TRIDENT_VERSION, + cli::{self, ClientArgs, ClientCommands, TridentExitCodes}, + command_name, + logging::operation_context, + run_command_if, ExitKind, OperationSource, TRIDENT_VERSION, }; -use crate::cli; - mod error; mod tridentclient; +use error::TridentClientError; use tridentclient::{RebootHandling, TridentClient}; pub fn client_main(args: &ClientArgs) -> ExitCode { @@ -24,7 +28,68 @@ pub fn client_main(args: &ClientArgs) -> ExitCode { return TridentExitCodes::SetupFailed.into(); }; - match runtime.block_on(run_client(args)) { + // `setup_tracing()` (see `main.rs`) treats grpc-client the same as any + // other command -- it's a first-class telemetry participant, not just + // a transport-only escape hatch, so it fires `command_start` like + // every other command. `command_error` is more selective (see + // `is_transport_failure` below). `run_client`'s errors are plain + // `anyhow::Error` (not `TridentError`), so they're wrapped in a + // generic `InternalError::Internal` here purely to get them into + // `run_command_if`'s `Result<_, TridentError>` shape -- the original + // anyhow context chain is preserved as the error's source and still + // printed in full below. + // Install/Update get the same stage/finalize-granular naming + // (`install_stage`, `update_finalize`, etc.) the CLI and daemon use for + // servicing telemetry -- otherwise every grpc-client update/install + // would collapse to the generic `client_update`/`client_install` + // regardless of which operations were actually requested. + let command = match &args.command { + ClientCommands::Install { + allowed_operations, .. + } + | ClientCommands::Update { + allowed_operations, .. + } => command_name( + args.command.name().trim_start_matches("client-"), + &cli::to_operations(allowed_operations), + ), + // Every other variant's name() is also "client-"-prefixed (see + // `ClientCommands::name()`) -- strip it here too so e.g. `commit`/ + // `stream_disk` match the CLI/daemon's own naming for the same + // logical command instead of reporting as `client_commit`/ + // `client_stream_disk`. + _ => args + .command + .name() + .trim_start_matches("client-") + .replace('-', "_"), + }; + + // `run_client` (the actual RPC) runs *inside* this closure, not before + // it, so `command_start` (fired by `run_command_if` the moment this + // closure is entered) actually brackets the RPC instead of always + // following it -- otherwise every client-side event the RPC itself + // fires, and the timestamp of `command_start` itself, would be + // reported after the call had already finished. `is_transport_failure` + // is computed from the raw `anyhow::Error` chain here, inside the + // closure, and stashed via `transport_failure` for `run_command_if`'s + // `should_report` predicate below, which only ever sees the already- + // wrapped `TridentError` and has no way to inspect that chain itself. + let transport_failure = Cell::new(false); + let result = run_command_if( + &command, + OperationSource::GrpcClient, + || { + let client_result = runtime.block_on(run_client(args)); + transport_failure.set(is_transport_failure(&client_result)); + client_result.map_err(|e| { + TridentError::with_source(InternalError::Internal("grpc-client command failed"), e) + }) + }, + |_error| transport_failure.get() && is_servicing_client_command(&args.command), + ); + + match result { Err(e) => { error!("Client failed: {:?}", e); return TridentExitCodes::Failed.into(); @@ -41,6 +106,68 @@ pub fn client_main(args: &ClientArgs) -> ExitCode { TridentExitCodes::Success.into() } +/// The daemon fires its own, correctly-classified `command_error` for any +/// request it actually received and acted on -- including one it rejected +/// outright (see e.g. `services::reject_invalid_argument`). A genuine +/// transport-level failure -- the daemon never received or finished +/// answering this request at all -- has no other reporter, and is what +/// this checks for: +/// - `ConnectionError`: the initial connection attempt itself failed +/// (socket not found, connection refused). +/// - `RequestError`/`ResponseError` whose wrapped `Status` is +/// `Code::Unavailable`, *except* for the daemon's own admission-control +/// rejections -- connection-lock or servicing-lock contention (see +/// `try_acquire_read_lock`/`try_acquire_write_lock`/`servicing_request`/ +/// `reading_request` in `server::tridentserver`), which deliberately +/// also return `Code::Unavailable` since a busy daemon is retryable the +/// same way a broken connection is. Those are recognized by their fixed +/// message text (`CONNECTION_LOCK_BUSY_MESSAGE`/`SERVICING_LOCK_BUSY_MESSAGE`) +/// and excluded here, since the daemon DID receive and answer this +/// request, unlike tonic's own `Code::Unavailable` for a connection that +/// broke mid-call (e.g. the daemon process died or the socket was closed +/// while a request/response was in flight). +fn is_transport_failure(client_result: &Result) -> bool { + client_result.as_ref().err().is_some_and(|e| { + e.chain() + .any(|cause| match cause.downcast_ref::() { + Some(TridentClientError::ConnectionError(..)) => true, + Some(TridentClientError::RequestError(_, status)) + | Some(TridentClientError::ResponseError(_, status)) => { + status.code() == Code::Unavailable + && status.message() != operation_context::CONNECTION_LOCK_BUSY_MESSAGE + && status.message() != operation_context::SERVICING_LOCK_BUSY_MESSAGE + } + _ => false, + }) + }) +} + +/// Whether `command` is a servicing operation for the purposes of the +/// `command_error` contract documented in `docs/Reference/Telemetry.md`'s +/// "Command Errors" section: only a servicing command's own failure gets a +/// `command_error` event -- `command_start` still fires for every command +/// (see the comment in `client_main` above). A read-only command like +/// `client-version` can still hit a transport-level failure (the daemon it +/// talked to was unreachable), but that failure isn't a "servicing command +/// failed" in the sense the docs describe, so it's deliberately excluded +/// here, mirroring the CLI's own read-only exclusions in `main.rs`. +/// +/// `Rollback { check: true, .. }` is `rollback --check`, a read-only dry +/// run (mirrors `main.rs`'s own `Commands::Rollback { check: true, .. }` +/// special-casing) -- only a real (non-check) rollback is a servicing +/// command here. +fn is_servicing_client_command(command: &ClientCommands) -> bool { + matches!( + command, + ClientCommands::Install { .. } + | ClientCommands::Update { .. } + | ClientCommands::Commit + | ClientCommands::RebuildRaid { .. } + | ClientCommands::Rollback { check: false, .. } + | ClientCommands::StreamDisk { .. } + ) +} + async fn run_client(args: &ClientArgs) -> Result { let mut client = TridentClient::connect(&args.server) .await diff --git a/crates/trident/src/lib.rs b/crates/trident/src/lib.rs index 328c9b77a..7a65eef9e 100644 --- a/crates/trident/src/lib.rs +++ b/crates/trident/src/lib.rs @@ -62,8 +62,9 @@ pub use crate::{ logfwd::LogForwarder, logstream::Logstream, operation_context::{ - run_with_captured_operation, run_with_operation, save_reboot_operation, - take_reboot_operation, OperationSource, + command_name, run_command, run_command_if, run_reboot_command, + run_with_captured_operation, save_reboot_operation, take_reboot_operation, + OperationSource, }, tracestream::TraceStream, }, @@ -313,7 +314,40 @@ impl Trident { )); } - tracing::info!(metric_name = "trident_start"); + // Best-effort: a failure to determine whether this is a CIH (Azure + // Container Linux) host must never fail startup, it only means + // this one field is missing from the trident_start telemetry. + // A detection failure is reported as "unknown", not "false" -- + // conflating "known non-ACL" with "couldn't tell" would + // misclassify a host whose CIH check simply failed to run as + // definitively non-ACL. + let acl = match cih::is_cih() { + Ok(true) => "true", + Ok(false) => "false", + Err(e) => { + warn!("Failed to determine if host is running CIH: {e:?}"); + "unknown" + } + }; + // CLOCK_BOOTTIME gives nanosecond-resolution time since boot + // (including any suspended time), unlike sysinfo::System::uptime() + // (or a naive /proc/uptime parse), which only exposes whole-second + // resolution. Best-effort: clock_gettime with a valid clock ID + // essentially never fails on Linux, but fall back to NaN (which + // serde_json serializes as JSON `null`, a genuine "not available" + // rather than a misleading literal zero) rather than failing + // startup if it somehow does. + let uptime_secs = nix::time::clock_gettime(nix::time::ClockId::CLOCK_BOOTTIME) + .map(|ts| Duration::from(ts).as_secs_f64()) + .unwrap_or_else(|e| { + warn!("Failed to read CLOCK_BOOTTIME: {e}"); + f64::NAN + }); + tracing::info!( + metric_name = "trident_start", + acl = acl, + uptime_secs = uptime_secs, + ); Ok(Self { host_config, @@ -875,7 +909,17 @@ impl Trident { self.host_config = Some(config); self.is_stream_image = true; - self.install(datastore, Operations::all(), false, Some(image)) + // `stream_image_start` above marks the beginning of a streamed + // install; mirror it with a completion signal here so streaming + // failures/successes are distinguishable in telemetry without + // relying on the downstream `clean_install_*` metrics (which are + // specific to the clean-install engine step, not the streaming + // entry point as a whole). + let result = self.install(datastore, Operations::all(), false, Some(image)); + if result.is_ok() { + tracing::info!(metric_name = "stream_image_success", value = true); + } + result } pub fn commit( diff --git a/crates/trident/src/logging/operation_context.rs b/crates/trident/src/logging/operation_context.rs index 71d2103fc..50477a4fb 100644 --- a/crates/trident/src/logging/operation_context.rs +++ b/crates/trident/src/logging/operation_context.rs @@ -24,10 +24,48 @@ //! `DataStore::correlation_id`, which is unrelated and set separately on //! `TraceStream`/`AppInsightsSender`). -use std::{cell::RefCell, sync::Mutex}; +use std::{ + cell::RefCell, + panic::{self, AssertUnwindSafe}, + sync::Mutex, +}; use uuid::Uuid; +use trident_api::{config::Operations, error::TridentError}; + +/// 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 +/// `"install_finalize"`), so `command`/`operation_id` telemetry is +/// consistent regardless of whether the command came from the CLI, from +/// gRPC/daemon, or was relayed through `grpc-client`. +pub fn command_name(base: &str, ops: &Operations) -> String { + match (ops.has_stage(), ops.has_finalize()) { + (true, true) => base.to_string(), + (true, false) => format!("{base}_stage"), + (false, true) => format!("{base}_finalize"), + // Neither stage nor finalize was requested (an empty + // `--allowed-operations` list); nothing can actually be staged or + // finalized, so name this like the existing no-op naming convention + // rather than a full install/update. + (false, false) => format!("{base}_noop"), + } +} + /// Identifies which of Trident's three entry points actually executed a /// command, so telemetry consumers can distinguish (for example) a /// `grpc-client` invocation that never reached a daemon from the daemon @@ -59,6 +97,17 @@ impl OperationSource { thread_local! { static CURRENT_OPERATION: RefCell> = const { RefCell::new(None) }; + + /// Set once [`report_command_error`] has already fired for the + /// current operation, so a domain-specific caller that needs to + /// explicitly emit `command_error` *before* some other side effect + /// (e.g. archiving the metrics file -- see + /// `engine::runtime_update::finalize_or_rollback_runtime_update`'s + /// auto-rollback failure path) doesn't cause `run_command`/ + /// `run_command_if` to report the same failure a second time once + /// that same error reaches their own `Err` handling. Reset alongside + /// `CURRENT_OPERATION` at the end of every command. + static COMMAND_ERROR_REPORTED: RefCell = const { RefCell::new(false) }; } /// Runs `f` with this thread tagged as executing `command` from `source`, @@ -91,6 +140,7 @@ pub fn run_with_operation(command: &str, source: OperationSource, f: impl FnO impl Drop for ClearOnDrop { fn drop(&mut self) { CURRENT_OPERATION.with(|cell| *cell.borrow_mut() = None); + COMMAND_ERROR_REPORTED.with(|cell| *cell.borrow_mut() = false); } } let _clear = ClearOnDrop; @@ -137,22 +187,30 @@ pub fn run_with_captured_operation( captured: Option, f: impl FnOnce() -> R, ) -> R { - let Some(CapturedOperation(operation_id, command, source)) = captured else { - return f(); - }; - - CURRENT_OPERATION.with(|cell| { - *cell.borrow_mut() = Some((operation_id, command, source)); - }); - + // Reset the dedup flag on entry and restore it on exit (even on + // panic), regardless of whether a context was actually captured -- + // this thread may be a reused `spawn_blocking` worker that last ran + // `run_reboot_command`/`report_command_error` and left the flag set, + // which would otherwise cause the *next* command on this thread to + // silently skip its own, unrelated `command_error`. struct ClearOnDrop; impl Drop for ClearOnDrop { fn drop(&mut self) { CURRENT_OPERATION.with(|cell| *cell.borrow_mut() = None); + COMMAND_ERROR_REPORTED.with(|cell| *cell.borrow_mut() = false); } } + COMMAND_ERROR_REPORTED.with(|cell| *cell.borrow_mut() = false); let _clear = ClearOnDrop; + let Some(CapturedOperation(operation_id, command, source)) = captured else { + return f(); + }; + + CURRENT_OPERATION.with(|cell| { + *cell.borrow_mut() = Some((operation_id, command, source)); + }); + f() } @@ -200,6 +258,147 @@ pub fn take_reboot_operation() -> Option { PENDING_REBOOT_OPERATION.lock().unwrap().take() } +/// Like [`run_command`], but specifically for the actual reboot call: +/// reuses whichever operation was captured by [`save_reboot_operation`] +/// (the servicing operation that decided a reboot was needed) instead of +/// minting a fresh `operation_id`, while still firing `command_error` on +/// failure -- so a failed reboot's error metric is correlated back to +/// that original operation, exactly like every other command's +/// `command_error`. Falls back to a plain, untagged call (still firing +/// `command_error` on failure) if nothing was captured. +pub fn run_reboot_command( + f: impl FnOnce() -> Result, +) -> Result { + run_with_captured_operation(take_reboot_operation(), || { + let result = f(); + if let Err(ref error) = result { + report_command_error(error); + } + result + }) +} + +/// Like [`run_with_operation`], but specifically for the +/// `Result` shape both places that run a command +/// actually use (CLI dispatch, gRPC's `servicing_request`/ +/// `reading_request`): additionally fires a `command_error` metric -- +/// breaking the error down into `kind`, `subkind`, and `location` -- if +/// `f` returns `Err`, while the operation_id/command context is still +/// active (so it's correlated the same way `command_start` is). +/// +/// Also catches a panic from `f` here, still inside +/// `run_with_operation`'s scope, so `command_error` still fires even when +/// `f` panics instead of returning `Err`. Without this, a panic would +/// unwind straight through this closure and past `run_with_operation`'s +/// `ClearOnDrop` guard -- which clears the operation context *during* the +/// unwind, before any code downstream of `run_command` (the CLI's own +/// outer `catch_unwind` in `main.rs`, or the daemon's per-request +/// panic-to-`Status`/`Completed` conversion) gets a chance to report +/// anything -- silently skipping `command_error` even though +/// `Telemetry.md` documents it as firing on every failed command. The +/// panic is re-raised afterwards via `resume_unwind`, so callers' +/// existing panic handling (exit codes, gRPC error responses) is +/// unaffected -- this only adds the metric emission that was missing. +pub fn run_command( + command: &str, + source: OperationSource, + f: impl FnOnce() -> Result, +) -> Result { + run_with_operation(command, source, || { + match panic::catch_unwind(AssertUnwindSafe(f)) { + Ok(result) => { + if let Err(ref error) = result { + report_command_error(error); + } + result + } + Err(payload) => { + report_command_panic(command, &payload); + panic::resume_unwind(payload); + } + } + }) +} + +/// Like [`run_command`], but only fires `command_error` when +/// `should_report` returns `true` for the resulting error. Use this when +/// the caller can prove some errors were already reported by someone else +/// (e.g. `grpc-client`, when the daemon it talked to already fired its +/// own, better-classified `command_error` for the same logical failure) +/// -- firing another one here would just double-count it under a less +/// informative classification. Panics are always reported regardless of +/// `should_report`: unlike an error from elsewhere, a panic has no other +/// reporter. +pub fn run_command_if( + command: &str, + source: OperationSource, + f: impl FnOnce() -> Result, + should_report: impl FnOnce(&TridentError) -> bool, +) -> Result { + run_with_operation(command, source, || { + match panic::catch_unwind(AssertUnwindSafe(f)) { + Ok(result) => { + if let Err(ref error) = result { + if should_report(error) { + report_command_error(error); + } + } + result + } + Err(payload) => { + report_command_panic(command, &payload); + panic::resume_unwind(payload); + } + } + }) +} + +/// Fires the `command_error` metric for a failed command. Split out from +/// `run_command` so it's independently testable against a constructed +/// `TridentError` without needing a real failing command. +/// +/// A no-op if `command_error` has already been reported once for the +/// current operation (see [`COMMAND_ERROR_REPORTED`]): a domain-specific +/// caller may need to fire this explicitly, ahead of some other side +/// effect that must observe the failure (e.g. archiving the metrics +/// file), before the same error naturally reaches `run_command`/ +/// `run_command_if`'s own `Err` handling further up the call stack -- +/// without this guard, that would report the identical failure twice. +pub(crate) fn report_command_error(error: &TridentError) { + if COMMAND_ERROR_REPORTED.with(|cell| cell.replace(true)) { + return; + } + tracing::info!( + metric_name = "command_error", + kind = error.kind().as_str(), + subkind = error.subkind().unwrap_or("none"), + location = error.location().as_str(), + ); +} + +/// Fires the `command_error` metric for a command that panicked instead +/// of returning `Err`. Tagged with a distinct `kind = "panic"` (rather +/// than reusing a `TridentError`'s own `kind`/`subkind`/`location`, which +/// don't exist for a panic) so consumers can tell the two failure modes +/// apart. The panic payload's message (when it is the common `&str` or +/// `String` panic message) is logged separately at `error` level, not +/// included in the metric's own fields, since it's unstructured and of +/// unbounded size. +fn report_command_panic(command: &str, payload: &Box) { + let message = payload + .downcast_ref::<&str>() + .map(|s| s.to_string()) + .or_else(|| payload.downcast_ref::().cloned()) + .unwrap_or_else(|| "".to_string()); + log::error!("Command '{command}' panicked: {message}"); + tracing::info!( + metric_name = "command_error", + kind = "panic", + subkind = "none", + location = "none", + ); +} + #[cfg(test)] mod tests { use super::*; @@ -367,4 +566,206 @@ mod tests { "nothing to capture outside an active operation" ); } + + #[test] + fn test_run_reboot_command_reuses_captured_operation() { + let _guard = REBOOT_OPERATION_TEST_LOCK.lock().unwrap(); + assert!( + take_reboot_operation().is_none(), + "start with a clean slate" + ); + + let expected = run_with_operation("install", OperationSource::Cli, || { + save_reboot_operation(); + current().unwrap() + }); + + let observed: Result<(String, String, OperationSource), TridentError> = + run_reboot_command(|| Ok(current().unwrap())); + assert_eq!( + observed.unwrap(), + expected, + "run_reboot_command should reuse the captured install operation, not mint a fresh one" + ); + } + + #[test] + fn test_run_reboot_command_fires_command_error_on_failure_even_without_capture() { + let _guard = REBOOT_OPERATION_TEST_LOCK.lock().unwrap(); + assert!( + take_reboot_operation().is_none(), + "start with a clean slate" + ); + + use tracing_subscriber::layer::SubscriberExt; + let layer = CapturingLayer::default(); + let events = layer.events.clone(); + let _guard2 = + tracing::subscriber::set_default(tracing_subscriber::Registry::default().with(layer)); + + let _: Result<(), TridentError> = + run_reboot_command(|| Err(TridentError::internal("reboot boom"))); + + let events = events.lock().unwrap(); + assert!( + events + .iter() + .any(|e| e.get("metric_name").map(String::as_str) == Some("command_error")), + "run_reboot_command should still fire command_error when nothing was captured" + ); + } + + #[test] + fn test_run_command_passes_through_ok() { + let result: Result = run_command("cmd", OperationSource::Cli, || Ok(42)); + assert_eq!(result.unwrap(), 42); + } + + #[test] + fn test_run_command_passes_through_err_unchanged() { + let result: Result<(), TridentError> = run_command("cmd", OperationSource::Cli, || { + Err(TridentError::internal("boom")) + }); + assert!(result.is_err()); + } + + #[test] + fn test_run_command_clears_context_after_error() { + let _: Result<(), TridentError> = run_command("cmd", OperationSource::Cli, || { + Err(TridentError::internal("boom")) + }); + assert!( + current().is_none(), + "context must be cleared even when f returns Err" + ); + } + + #[test] + fn test_run_command_if_suppresses_report_when_predicate_false() { + use tracing_subscriber::layer::SubscriberExt; + + let layer = CapturingLayer::default(); + let events = layer.events.clone(); + let _guard = + tracing::subscriber::set_default(tracing_subscriber::Registry::default().with(layer)); + + let result: Result<(), TridentError> = run_command_if( + "cmd", + OperationSource::GrpcClient, + || Err(TridentError::internal("boom")), + |_error| false, + ); + + assert!(result.is_err()); + let events = events.lock().unwrap(); + assert!( + !events + .iter() + .any(|e| e.get("metric_name").map(String::as_str) == Some("command_error")), + "command_error should not fire when should_report returns false: {events:?}" + ); + } + + #[test] + fn test_run_command_if_reports_when_predicate_true() { + use tracing_subscriber::layer::SubscriberExt; + + let layer = CapturingLayer::default(); + let events = layer.events.clone(); + let _guard = + tracing::subscriber::set_default(tracing_subscriber::Registry::default().with(layer)); + + let result: Result<(), TridentError> = run_command_if( + "cmd", + OperationSource::GrpcClient, + || Err(TridentError::internal("boom")), + |_error| true, + ); + + assert!(result.is_err()); + let events = events.lock().unwrap(); + assert!( + events + .iter() + .any(|e| e.get("metric_name").map(String::as_str) == Some("command_error")), + "command_error should fire when should_report returns true: {events:?}" + ); + } + + /// A minimal `tracing_subscriber::Layer` that records every event's + /// fields as strings, so `report_command_error`'s output can be + /// asserted on directly instead of only checking that `run_command` + /// doesn't panic. + #[derive(Default, Clone)] + struct CapturingLayer { + events: std::sync::Arc>>>, + } + + struct CaptureVisitor(std::collections::BTreeMap); + + impl tracing::field::Visit for CaptureVisitor { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + self.0.insert(field.name().to_string(), value.to_string()); + } + + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + self.0 + .insert(field.name().to_string(), format!("{value:?}")); + } + } + + impl tracing_subscriber::layer::Layer for CapturingLayer + where + S: tracing::Subscriber, + { + fn on_event( + &self, + event: &tracing::Event<'_>, + _ctx: tracing_subscriber::layer::Context<'_, S>, + ) { + let mut visitor = CaptureVisitor(std::collections::BTreeMap::new()); + event.record(&mut visitor); + self.events.lock().unwrap().push(visitor.0); + } + } + + #[test] + fn test_run_command_fires_command_error_with_kind_subkind_location() { + use tracing_subscriber::layer::SubscriberExt; + + let layer = CapturingLayer::default(); + let events = layer.events.clone(); + let _guard = + tracing::subscriber::set_default(tracing_subscriber::Registry::default().with(layer)); + + let _: Result<(), TridentError> = run_command("test_command", OperationSource::Cli, || { + Err(TridentError::internal("boom")) + }); + + let events = events.lock().unwrap(); + let command_error = events + .iter() + .find(|e| e.get("metric_name").map(String::as_str) == Some("command_error")) + .expect("command_error event should have been fired"); + + assert_eq!( + command_error.get("kind").map(String::as_str), + Some("internal") + ); + assert!( + command_error.get("subkind").is_some(), + "subkind should be present: {command_error:?}" + ); + assert!( + command_error + .get("location") + .is_some_and(|l| l.contains("operation_context.rs")), + "location should point at the TridentError::internal call site: {command_error:?}" + ); + // command_start (from run_with_operation) should also have fired, + // ahead of command_error. + assert!(events + .iter() + .any(|e| e.get("metric_name").map(String::as_str) == Some("command_start"))); + } } diff --git a/crates/trident/src/logging/tracestream.rs b/crates/trident/src/logging/tracestream.rs index 531128f49..2ee032c73 100644 --- a/crates/trident/src/logging/tracestream.rs +++ b/crates/trident/src/logging/tracestream.rs @@ -1,6 +1,6 @@ use std::{ collections::BTreeMap, - fs::{self, File}, + fs::{self, File, OpenOptions}, io::Write, path::Path, sync::{Arc, RwLock}, @@ -22,7 +22,6 @@ use tracing_subscriber::{layer::Layer, registry::LookupSpan}; use trident_api::error::TridentError; use osutils::{ - files, osrelease::{OsRelease, OS_RELEASE_PATH}, uname, }; @@ -288,26 +287,42 @@ impl TraceStream { Ok(()) } - /// Create a Boxed TraceSender + /// Create a Boxed TraceSender. Truncates the local metrics file on + /// creation, same as every previous invocation of a command that + /// installs this layer -- appropriate for commands that are + /// themselves generating fresh servicing metrics. pub fn make_trace_sender(&self) -> Box { - self.make_trace_sender_with_metrics_path(TRIDENT_METRICS_FILE_PATH) + self.make_trace_sender_with_metrics_path(TRIDENT_METRICS_FILE_PATH, true) + } + + /// Like `make_trace_sender`, but appends to the existing local metrics + /// file instead of truncating it. For commands (namely `diagnose`) + /// that read back and repackage that same file's *pre-existing* + /// content (e.g. into a support bundle) -- truncating it first would + /// destroy the history the command is supposed to be collecting, + /// leaving only the metrics the command emits about itself. + pub fn make_trace_sender_appending(&self) -> Box { + self.make_trace_sender_with_metrics_path(TRIDENT_METRICS_FILE_PATH, false) } /// Like `make_trace_sender`, but writes the local metrics file to /// `metrics_file_path` instead of the real host path - /// (`TRIDENT_METRICS_FILE_PATH`). This lets tests exercise the full + /// (`TRIDENT_METRICS_FILE_PATH`), and lets the caller choose whether + /// to truncate it first. This lets tests exercise the full /// metrics-writing pipeline against a throwaway temp file instead of a /// real, shared host path, so they can be plain `#[test]`s instead of /// needing a VM. pub(crate) fn make_trace_sender_with_metrics_path( &self, metrics_file_path: &str, + truncate: bool, ) -> Box { Box::new(TraceSender::new( self.target.clone(), self.installation_id.clone(), self.datastore_id.clone(), metrics_file_path, + truncate, )) } } @@ -332,13 +347,45 @@ impl TraceSender { installation_id: Arc>>, datastore_id: Arc>>, metrics_file_path: &str, + truncate: bool, ) -> Self { + if let Some(parent) = Path::new(metrics_file_path).parent() { + if let Err(err) = fs::create_dir_all(parent) { + eprintln!( + "Tracestream setup error: failed to create local metrics file's parent directory: {err:?}" + ); + } + } + // Reset any pre-existing content up front when requested, via a + // separate truncating open, then always keep the real handle in + // append-only mode: a plain `File::create` (O_TRUNC without + // O_APPEND) kept open long-term has its own independent, + // non-advancing write offset, so a concurrent writer to this same + // path (e.g. `grpc-client`, opened separately in append mode) that + // extends the file past that offset would have its data + // overwritten the next time this descriptor writes. Combining + // `OpenOptions::truncate(true)` with `.append(true)` in one open() + // call isn't an option: the standard library requires `.write(true)` + // for truncation, and adding that back defeats the point of + // append-only semantics for every later write through this handle. + if truncate { + if let Err(err) = File::create(metrics_file_path) { + eprintln!( + "Tracestream setup error: failed to truncate local metrics file: {err:?}" + ); + } + } + let metrics_file = OpenOptions::new() + .create(true) + .append(true) + .open(metrics_file_path) + .map_err(Error::from); Self { server, installation_id, datastore_id, client: reqwest::blocking::Client::new(), - metrics_file: match files::create_file(metrics_file_path) { + metrics_file: match metrics_file { Ok(f) => Some(f), Err(err) => { eprintln!( @@ -694,7 +741,7 @@ mod tests { let metrics_path = temp_dir.path().join("metrics.jsonl"); let tracestream = TraceStream::default(); let trace_sender = - tracestream.make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap()); + tracestream.make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap(), true); assert!( trace_sender.get_server().is_none(), "tracestream should not have a server" @@ -711,13 +758,52 @@ mod tests { ); } + #[test] + /// Regression test: `make_trace_sender_with_metrics_path(.., false)` + /// (used by `make_trace_sender_appending`, for `diagnose`) must append + /// to a pre-existing metrics file rather than truncating it -- unlike + /// the `true` (truncating) case every other command uses. + fn test_tracestream_appending_sender_preserves_existing_metrics() { + let temp_dir = tempfile::tempdir().unwrap(); + let metrics_path = temp_dir.path().join("metrics.jsonl"); + std::fs::write(&metrics_path, "preexisting line\n").unwrap(); + + let tracestream = TraceStream::default(); + let trace_sender = tracestream + .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap(), false) + .with_filter(filter::LevelFilter::INFO); + + let _guard = tracing::subscriber::set_default( + tracing_subscriber::Registry::default().with(trace_sender), + ); + + tracing::info!(metric_name = "test_metric_appended", value = true); + + std::thread::sleep(std::time::Duration::from_millis(100)); + + let file = File::open(&metrics_path).unwrap(); + let reader = BufReader::new(file); + let lines: Vec = reader.lines().map(|l| l.unwrap()).collect(); + + assert!( + lines.iter().any(|line| line == "preexisting line"), + "appending sender must not have truncated the pre-existing content" + ); + assert!( + lines + .iter() + .any(|line| line.contains(r#""metric_name":"test_metric_appended""#)), + "appending sender must still write new metrics" + ); + } + #[test] fn test_lock() { let temp_dir = tempfile::tempdir().unwrap(); let metrics_path = temp_dir.path().join("metrics.jsonl"); let mut tracestream = TraceStream::default(); let trace_sender = - tracestream.make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap()); + tracestream.make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap(), true); assert!( trace_sender.get_server().is_none(), @@ -758,7 +844,7 @@ mod tests { let metrics_path = temp_dir.path().join("metrics.jsonl"); let tracestream = TraceStream::default(); let trace_sender = tracestream - .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap()) + .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap(), true) .with_filter(filter::LevelFilter::INFO); // Use a thread-local scoped default subscriber (rather than @@ -802,7 +888,7 @@ mod tests { let tracestream = TraceStream::default(); tracestream.set_installation_id("test-installation-id".to_string()); let trace_sender = tracestream - .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap()) + .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap(), true) .with_filter(filter::LevelFilter::INFO); // See test_tracestream_write_metric_event_to_file for why a scoped @@ -844,7 +930,7 @@ mod tests { let tracestream = TraceStream::default(); tracestream.set_datastore_id("test-datastore-id".to_string()); let trace_sender = tracestream - .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap()) + .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap(), true) .with_filter(filter::LevelFilter::INFO); // See test_tracestream_write_metric_event_to_file for why a scoped @@ -879,7 +965,7 @@ mod tests { let metrics_path = temp_dir.path().join("metrics.jsonl"); let tracestream = TraceStream::default(); let trace_sender = tracestream - .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap()) + .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap(), true) .with_filter(filter::LevelFilter::INFO); // See test_tracestream_write_metric_event_to_file for why a scoped diff --git a/crates/trident/src/main.rs b/crates/trident/src/main.rs index 132c64480..e42755411 100644 --- a/crates/trident/src/main.rs +++ b/crates/trident/src/main.rs @@ -8,36 +8,18 @@ use osutils::logging::{filter::LogFilter, multilog::MultiLogger}; use trident::{ agentconfig::AgentConfig, cli::{self, Cli, Commands, GetKind, TridentExitCodes}, + command_name, init::offline, manual_rollback::{self, utils::ManualRollbackRequestKind}, - run_with_captured_operation, run_with_operation, save_reboot_operation, take_reboot_operation, - validation, AppInsightsSender, BackgroundLog, BackgroundUploader, DataStore, ExitKind, - LogForwarder, Logstream, OperationSource, TraceStream, Trident, TRIDENT_BACKGROUND_LOG_PATH, + run_command, run_reboot_command, save_reboot_operation, validation, AppInsightsSender, + BackgroundLog, BackgroundUploader, DataStore, ExitKind, LogForwarder, Logstream, + OperationSource, TraceStream, Trident, TRIDENT_BACKGROUND_LOG_PATH, }; use trident_api::{ - config::{HostConfigurationSource, Operations}, + config::HostConfigurationSource, error::{InternalError, InvalidInputError, TridentError, TridentResultExt}, }; -/// 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 -/// `"install_finalize"`), so `command`/`operation_id` telemetry is -/// consistent regardless of whether the command came from the CLI or from -/// gRPC/daemon. -fn command_name(base: &str, ops: &Operations) -> String { - match (ops.has_stage(), ops.has_finalize()) { - (true, true) => base.to_string(), - (true, false) => format!("{base}_stage"), - (false, true) => format!("{base}_finalize"), - // Neither stage nor finalize was requested (an empty - // `--allowed-operations` list); nothing can actually be staged or - // finalized, so name this like the existing no-op naming convention - // rather than a full install/update. - (false, false) => format!("{base}_noop"), - } -} - fn run_trident( mut logstream: Logstream, mut tracestream: TraceStream, @@ -67,7 +49,17 @@ fn run_trident( proxy_status("NO_PROXY"), ); - // Catch exit fast commands + // Fast-exit commands: read-only/one-shot commands that never start a + // servicing run (validate, get, diagnose, offline-initialize, a manual + // rollback --check, start-network). These deliberately run outside + // run_command below -- no command_start/command_error telemetry is + // emitted for them. Their failures (a malformed --config, a datastore + // that can't be opened, a diagnostics bundle that can't be written, + // etc.) are operator-input or read errors, not servicing outcomes; a + // genuine underlying datastore/host problem still gets telemetry when + // the actual servicing operation (install/update/etc.) that triggered + // it runs. Handled here, before `command`/run_command are even set up, + // so none of that machinery needs to reason about them. match &args.command { Commands::Validate { config } => { return validation::validate_host_config_file(config).map(|()| ExitKind::Done); @@ -141,90 +133,106 @@ fn run_trident( _ => (), } - let res = panic::catch_unwind(move || { - match &args.command { - Commands::Install { status, error, .. } - | Commands::Update { status, error, .. } - | Commands::Commit { status, error } - | Commands::RebuildRaid { status, error, .. } - | Commands::Rollback { status, error, .. } => { - // Determined before any preflight checks below, and used - // to wrap the *entire* servicing branch (preflight checks, - // Trident::new, and the actual command) in a single - // run_with_operation call -- not just the innermost - // install/update/commit/rollback/rebuild-raid call, as - // before. That previously left Trident::new (and - // everything it does, including firing "trident_start") - // outside any operation context: every event from CLI - // startup through to just before the actual command ran - // had no operation_id/command. - let command = match &args.command { - Commands::Install { - allowed_operations, .. - } => command_name(args.command.name(), &cli::to_operations(allowed_operations)), - Commands::Update { - allowed_operations, .. - } => command_name(args.command.name(), &cli::to_operations(allowed_operations)), - Commands::Commit { .. } => "commit".to_string(), - Commands::Rollback { - allowed_operations, .. - } => command_name(args.command.name(), &cli::to_operations(allowed_operations)), - Commands::RebuildRaid { .. } => "rebuild_raid".to_string(), - _ => unreachable!(), - }; - - // Determined up front, before any preflight checks below, - // so a missing/nonexistent --config is rejected immediately. - let config_path = match &args.command { - Commands::Update { config, .. } | Commands::Install { config, .. } => { - Some(config.clone()) - } - Commands::RebuildRaid { config, .. } => config.clone(), - _ => None, - }; - if let Some(path) = &config_path { - if !path.exists() { - return run_with_operation(&command, OperationSource::Cli, || { - Err(TridentError::new(InvalidInputError::ReadInputFile { - path: path.to_string_lossy().to_string(), - })) - .message("Config file does not exist") - }); - } - } + // Only servicing commands reach here: Install, Update, Commit, + // RebuildRaid, and a non-check Rollback. These get command_start/ + // command_error telemetry via run_command below; the fast-exit + // commands above already returned without any. + let command = match &args.command { + Commands::Install { + allowed_operations, .. + } => command_name(args.command.name(), &cli::to_operations(allowed_operations)), + Commands::Update { + allowed_operations, .. + } => command_name(args.command.name(), &cli::to_operations(allowed_operations)), + Commands::Rollback { + allowed_operations, .. + } => command_name(args.command.name(), &cli::to_operations(allowed_operations)), + Commands::Commit { .. } | Commands::RebuildRaid { .. } => { + args.command.name().replace('-', "_") + } + Commands::StartNetwork { .. } + | Commands::Get { .. } + | Commands::Diagnose { .. } + | Commands::Validate { .. } + | Commands::OfflineInitialize { .. } => { + unreachable!("fast-exit commands already returned above") + } + #[cfg(feature = "pytest-generator")] + Commands::Pytest => unreachable!("fast-exit commands already returned above"), + Commands::Daemon { .. } | Commands::GrpcClient(_) => { + unreachable!("Daemon/GrpcClient are dispatched in main(), never reach run_trident") + } + }; - // Attach this host's installation ID and database ID to - // the shared TraceStream before run_with_operation below - // fires command_start: Trident::new (further down, inside - // the closure) is the usual place both get attached, but - // that's too late for command_start, which - // run_with_operation fires immediately, before the closure - // even runs. Both are read-only and side-effect-free: - // neither creates a datastore or an ID (see - // `TraceStream::attach_installation_id_if_present` and - // `TraceStream::attach_datastore_id_if_present`) -- silently - // does nothing if the datastore doesn't exist yet, which is - // expected for a host's first-ever install. - // - // Load once and reuse the same snapshot for both the - // pre-warm attach here and the operation closure below: - // calling `AgentConfig::load()` a second time inside the - // closure could observe a different `DatastorePath` (e.g. a - // CIH bootstrap swap between the two reads), leaving the - // IDs cached on `tracestream` here attributed to a - // different datastore than the one the operation actually - // runs against. - let agent_config_result = AgentConfig::load(); - if let Ok(agent_config) = &agent_config_result { - tracestream.attach_installation_id_if_present(agent_config.datastore_path()); - tracestream.attach_datastore_id_if_present(agent_config.datastore_path()); - } + // Attach this host's installation ID and database ID to the shared + // TraceStream before run_command below fires command_start: + // Trident::new (further down, inside the closure) is the usual place + // both get attached, but that's too late for command_start, which + // run_command fires immediately, before the closure even runs. Both + // are read-only and side-effect-free: neither creates a datastore or + // an ID (see `TraceStream::attach_installation_id_if_present` and + // `TraceStream::attach_datastore_id_if_present`) -- silently does + // nothing if the datastore doesn't exist yet, which is expected for a + // host's first-ever install. + // Load once and reuse the same snapshot inside the run_command + // closure below, rather than reloading there: calling + // `AgentConfig::load()` a second time could observe a different + // `DatastorePath` (e.g. a CIH bootstrap swap between the two reads), + // leaving the IDs attached to `tracestream` here attributed to a + // different datastore than the one the operation actually runs + // against. + let agent_config_result = AgentConfig::load(); + if let Ok(agent_config) = &agent_config_result { + tracestream.attach_installation_id_if_present(agent_config.datastore_path()); + tracestream.attach_datastore_id_if_present(agent_config.datastore_path()); + } - run_with_operation(&command, OperationSource::Cli, || { + // Determined up front so a missing/nonexistent --config is rejected + // immediately, before run_command below even fires command_start. + let config_path = match &args.command { + Commands::Update { config, .. } | Commands::Install { config, .. } => Some(config.clone()), + Commands::RebuildRaid { config, .. } => config.clone(), + _ => None, + }; + if let Some(path) = &config_path { + if !path.exists() { + return run_command(&command, OperationSource::Cli, || { + Err(TridentError::new(InvalidInputError::ReadInputFile { + path: path.to_string_lossy().to_string(), + })) + .message("Config file does not exist") + }); + } + } + + // run_command itself now catches a panic from its closure (while the + // operation context is still active) and fires command_error before + // re-raising it, so a genuine panic gets the same telemetry as a + // normal Err. This outer catch_unwind remains as a safety net for a + // panic occurring outside run_command's closure (e.g. in run_command's + // own setup) and to keep converting an unwound panic into a non-zero + // exit code below. + let res = panic::catch_unwind(move || { + run_command(&command, OperationSource::Cli, || { + match &args.command { + Commands::Install { status, error, .. } + | Commands::Update { status, error, .. } + | Commands::Commit { status, error } + | Commands::RebuildRaid { status, error, .. } + | Commands::Rollback { status, error, .. } => { // config_path was already validated (existence-checked) - // above. Reuse the same `AgentConfig` snapshot loaded - // just above rather than reloading -- see the comment - // there. + // above. + let config_path = match &args.command { + Commands::Update { config, .. } | Commands::Install { config, .. } => { + Some(config.clone()) + } + Commands::RebuildRaid { config, .. } => config.clone(), + _ => None, + }; + + // Reuse the same `AgentConfig` snapshot loaded above + // (before this closure/panic::catch_unwind) rather + // than reloading -- see the comment there. let agent_config = agent_config_result?; // For commands that cannot themselves stage a new // install/update (see @@ -346,7 +354,7 @@ fn run_trident( // Capture this operation's identity while its context // is still installed (this closure runs entirely - // inside `run_with_operation`'s scope), so the reboot + // inside `run_command`'s scope), so the reboot // requested below by the caller can be tagged with the // *original* install/update/etc.'s `operation_id`/ // `command` instead of a disconnected fresh one -- see @@ -356,10 +364,10 @@ fn run_trident( } res.message(format!("Failed to execute '{}' command", args.command)) - }) + } + _ => unreachable!(), } - _ => unreachable!(), - } + }) }); match res { @@ -437,7 +445,11 @@ fn setup_logging( #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum TelemetryStatus { /// Tracing/telemetry setup does not apply to this command at all (the - /// `_ => {}` arm in [`setup_tracing`]) -- not logged. + /// `Commands::Pytest` arm in [`setup_tracing`], gated behind the + /// `pytest-generator` feature) -- not logged. Only ever constructed + /// when that feature is enabled; every other command now gets a real + /// subscriber installed. + #[cfg_attr(not(feature = "pytest-generator"), allow(dead_code))] NotApplicable, /// `Telemetry=OptOut` (the default): telemetry was never attempted. OptedOut, @@ -499,20 +511,84 @@ fn setup_tracing( use tracing_subscriber::{filter, layer::SubscriberExt, Layer, Registry}; let tracestream = TraceStream::default(); - let mut telemetry_status = TelemetryStatus::NotApplicable; - + let telemetry_status; + + // Every command reachable from run_trident needs a subscriber + // installed here -- not just the servicing ones -- so that ordinary + // logging (journald) and, for the servicing commands, command_start/ + // command_error all reach a real subscriber instead of the default + // one, which is none at all: tracing silently drops every event with + // no subscriber installed. The fast "exit early" commands (validate, + // get, diagnose, offline-initialize, rollback --check, start-network) + // don't emit command_start/command_error or any other metric_name + // event themselves (see run_trident), but still get a subscriber here + // -- see the local_sender truncate-vs-append comment below for why. + // StartNetwork's own `tracestream.disable()` (see run_trident) still + // applies regardless -- it only suppresses a later `set_server` call + // from configuring a remote phone-home target before the network + // exists, not the local metrics-file/journald layers installed here, + // which need no network. match &args.command { Commands::Commit { .. } | Commands::Daemon { .. } | Commands::GrpcClient { .. } | Commands::Install { .. } | Commands::RebuildRaid { .. } - | Commands::Rollback { check: false, .. } - | Commands::Update { .. } => { + | Commands::Rollback { .. } + | Commands::Update { .. } + | Commands::Validate { .. } + | Commands::Get { .. } + | Commands::Diagnose { .. } + | Commands::OfflineInitialize { .. } + | Commands::StartNetwork { .. } => { + // Truncating the local metrics file is only appropriate for a + // process that actually owns the file's lifecycle for a fresh + // servicing run -- Install/Update/Commit/RebuildRaid/Rollback + // (finalize)/Daemon -- since those are the operations whose + // metrics history is meaningful to reset per invocation. Every + // other command reads or inspects existing state without + // mutating it, so it must append instead of truncating -- not + // because any of them emit command_start/command_error or any + // other metric_name event of their own (they don't; see + // run_trident), but because simply *opening* the file with + // truncation is itself destructive: + // * `validate`, `get`, `diagnose`, `offline-initialize`, + // `start-network`, and a manual rollback `--check` are all + // read-only/fast commands that never start a servicing run -- + // truncating here would erase the preceding servicing + // metrics history just because one of these ran afterward. + // * `Commands::GrpcClient` -- *every* subcommand of it, not + // just its own read-only ones (`get`, `validate`, + // `rollback --check`) -- never owns this file either way: by + // definition it only ever talks to an *already-running* + // daemon, which is the file's sole owner for as long as it's + // up. That makes truncation from a grpc-client process + // incorrect unconditionally, including for + // install/update/rollback (finalize) subcommands: those + // still start a *servicing* run, but that run is owned and + // recorded by the daemon, not by the short-lived grpc-client + // process asking for it. Previously only the read-only + // grpc-client subcommands were special-cased here, so a + // `grpc-client install`/`update`/`rollback` truncated the + // shared local metrics file out from under the daemon's own + // concurrent appends -- the exact race this append-mode + // split was meant to prevent. + let local_sender = if matches!( + args.command, + Commands::GrpcClient(_) + | Commands::Diagnose { .. } + | Commands::Validate { .. } + | Commands::Get { .. } + | Commands::OfflineInitialize { .. } + | Commands::StartNetwork { .. } + | Commands::Rollback { check: true, .. } + ) { + tracestream.make_trace_sender_appending() + } else { + tracestream.make_trace_sender() + }; let mut layers: Vec + Send + Sync>> = vec![Box::new( - tracestream - .make_trace_sender() - .with_filter(filter::LevelFilter::INFO), + local_sender.with_filter(filter::LevelFilter::INFO), )]; // As functionality moves to the Daemon, move the journald layer to @@ -565,8 +641,16 @@ fn setup_tracing( tracing::subscriber::set_global_default(Registry::default().with(layers)) .context("Failed to set global default subscriber")?; } - _ => { - // no op + // pytest-generator does no meaningful work of its own (just + // generates functional-test wrappers at build/dev time) -- no + // telemetry needed. Listed explicitly, rather than via a wildcard + // fallback, so the compiler forces this match to be revisited + // whenever a new command variant is added, instead of it silently + // falling through to "no subscriber" the way the commands above + // used to. + #[cfg(feature = "pytest-generator")] + Commands::Pytest => { + telemetry_status = TelemetryStatus::NotApplicable; } } @@ -718,19 +802,18 @@ fn main() -> ExitCode { Ok(ExitKind::NeedsReboot) => { // Reuse the just-completed install/update/etc.'s own // operation_id/command (captured via save_reboot_operation - // just before that command's own run_with_operation scope - // ended) rather than leaving `trident_system_reboot` - // untagged, or minting an unrelated fresh "reboot" - // identity: the reboot is a direct continuation of that - // same servicing operation, not an independent one, so - // telemetry should correlate it back to the same - // operation_id. Falls back to a plain, untagged call if - // nothing was captured (shouldn't happen on this path, but - // avoids losing the reboot attempt entirely if it does). - if let Err(e) = run_with_captured_operation( - take_reboot_operation(), - trident::request_reboot_with_wait, - ) { + // just before that command's own run_command scope ended) + // rather than leaving `trident_system_reboot` untagged, or + // minting an unrelated fresh "reboot" identity: the reboot + // is a direct continuation of that same servicing + // operation, not an independent one, so telemetry should + // correlate it back to the same operation_id. + // run_reboot_command also still fires command_error on a + // failed reboot -- falling back to a fresh, untagged + // command if nothing was captured -- matching every other + // command's error-reporting contract instead of silently + // dropping this one on the floor. + if let Err(e) = run_reboot_command(trident::request_reboot_with_wait) { error!("Failed to reboot: {e:?}"); return TridentExitCodes::RebootUnsuccessful.into(); } diff --git a/crates/trident/src/server/mod.rs b/crates/trident/src/server/mod.rs index 8e7b01dbd..40dbe769e 100644 --- a/crates/trident/src/server/mod.rs +++ b/crates/trident/src/server/mod.rs @@ -32,6 +32,8 @@ use trident_proto::v1preview::{ status_service_server::StatusServiceServer, validation_service_server::ValidationServiceServer, }; +use trident_api::error::{ReportError, ServicingError, TridentError}; + use crate::{ agentconfig::AgentConfig, cli::TridentExitCodes, @@ -174,21 +176,25 @@ fn reboot(signals: ShutdownSignals) -> ExitCode { // time this function runs, the whole daemon event loop // (`server_main_inner`/`main_task`) has already returned, so there is // no thread-local operation context active here at all. - let request_result = operation_context::run_with_captured_operation( - operation_context::take_reboot_operation(), - reboot::request_reboot, - ); - if let Err(e) = request_result { - error!("Failed to request reboot: {e:?}"); - return TridentExitCodes::RebootUnsuccessful.into(); - } + // + // Routed through `run_reboot_command` (not + // `run_with_captured_operation` directly) so a failed reboot fires + // `command_error` here exactly like the CLI's reboot path does, + // making a failed daemon-driven reboot (systemctl unreachable, or the + // shutdown signal never arriving) distinguishable in telemetry from a + // successful one. + let reboot_result: Result<(), TridentError> = operation_context::run_reboot_command(|| { + reboot::request_reboot().structured(ServicingError::Reboot)?; + + // Wait for either a shutdown signal or the reboot timeout. + signals + .exit_receiver() + .recv_timeout(Duration::from_secs(REBOOT_WAIT_DURATION_SECS)) + .structured(ServicingError::RebootTimeout) + }); - // Wait for either a shutdown signal or the reboot timeout - if let Err(e) = signals - .exit_receiver() - .recv_timeout(Duration::from_secs(REBOOT_WAIT_DURATION_SECS)) - { - error!("Reboot wait timed out: {e:?}"); + if let Err(e) = reboot_result { + error!("Failed to reboot: {e:?}"); return TridentExitCodes::RebootUnsuccessful.into(); } diff --git a/crates/trident/src/server/tridentserver/mod.rs b/crates/trident/src/server/tridentserver/mod.rs index d3a60a4c2..a0cbfd3d3 100644 --- a/crates/trident/src/server/tridentserver/mod.rs +++ b/crates/trident/src/server/tridentserver/mod.rs @@ -160,40 +160,59 @@ 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) }) } - /// Re-attaches a persisted installation ID and datastore ID to - /// `self.tracestream`, if either is now available but wasn't at - /// daemon-startup time (`server_main`'s one-time attach runs before any - /// request has had a chance to create a datastore, so a request that - /// arrives before the very first install/update -- and whose own - /// handler goes on to create that datastore -- would otherwise still be - /// missing both IDs. Uses `self.agent_config` (the same configuration - /// the request itself operates on) rather than reloading from disk, so - /// this can't refresh from a different datastore path than the one in - /// effect for this request, and a transient reload failure can't - /// silently skip the refresh. Neither call creates a datastore: both - /// silently do nothing if the datastore doesn't exist yet. But on an - /// existing datastore, either call may still *persist* a missing ID -- + /// Re-checks for a persisted installation ID and database ID before a + /// request fires its own `command_start` (via `run_command`). The + /// daemon-startup attach in `server_main` only ever runs once, at + /// startup -- so a daemon that starts before the host is ever + /// installed, then serves a request some time after a *different* + /// path (e.g. a concurrent CLI invocation, or an earlier servicing + /// request on this same daemon) has since created the datastore, + /// would otherwise still be missing both IDs. Called from both + /// `servicing_request` and `reading_request`, so read-only RPCs (e.g. + /// `get_servicing_state`) don't keep reporting missing IDs + /// indefinitely just because they never happen to run after a write + /// request has attached them. Uses `self.agent_config` (the same + /// configuration the request itself operates on) rather than + /// reloading from disk, so this can't refresh from a different + /// datastore path than the one in effect for this request, and a + /// transient reload failure can't silently skip the refresh. Neither + /// call creates a datastore: both silently do nothing if the + /// datastore doesn't exist yet. But on an existing datastore, either + /// call may still *persist* a missing ID -- /// `attach_datastore_id_if_present` via `DataStore::datastore_id`'s - /// get-or-create semantics, and `attach_installation_id_if_present` via - /// `DataStore::installation_id_or_migrate`'s legacy-ID migration (see + /// get-or-create semantics, and `attach_installation_id_if_present` + /// via `DataStore::installation_id_or_migrate`'s legacy-ID migration + /// (see /// `TraceStream::attach_installation_id_if_present` and /// `TraceStream::attach_datastore_id_if_present`). fn refresh_ids(&self) { @@ -271,26 +290,30 @@ impl TridentServer { // can tag `trident_system_reboot` with this same servicing // operation's identity instead of leaving it untagged. let f = move || { - operation_context::run_with_operation( - name, - operation_context::OperationSource::Daemon, - || { - let result = f(); - if let Ok((ExitKind::NeedsReboot, ..)) = &result { - operation_context::save_reboot_operation(); - } - result - }, - ) + operation_context::run_command(name, operation_context::OperationSource::Daemon, || { + 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", name); - 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. @@ -387,15 +410,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) diff --git a/crates/trident/src/server/tridentserver/services/install.rs b/crates/trident/src/server/tridentserver/services/install.rs index 15aed3cac..553dbfaf2 100644 --- a/crates/trident/src/server/tridentserver/services/install.rs +++ b/crates/trident/src/server/tridentserver/services/install.rs @@ -26,17 +26,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( + "install", + "stage", + "Missing staging configuration", + )); }; let Some(host_config) = staging.config else { - return Err(Status::invalid_argument( + return Err(self.reject_invalid_argument( + "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( + "install", + "finalize", + "Missing finalize configuration", + )); }; let data_store_path = self.agent_config.datastore_path().to_owned(); @@ -73,7 +83,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( + "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 e58bd12c6..23e612ab3 100644 --- a/crates/trident/src/server/tridentserver/services/mod.rs +++ b/crates/trident/src/server/tridentserver/services/mod.rs @@ -1,6 +1,11 @@ +use tonic::Status; +use trident_api::error::{InvalidInputError, TridentError}; use trident_proto::v1::{RebootHandling, RebootManagement}; -use crate::server::tridentserver::RebootDecision; +use crate::{ + logging::operation_context, + server::tridentserver::{RebootDecision, TridentServer}, +}; mod commit; mod rollback; @@ -17,6 +22,96 @@ 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. + /// + /// Calls `refresh_ids` first, exactly like + /// `servicing_request`/`reading_request` do, so a rejected request + /// still gets the host's installation ID attached when one is + /// available. Without this, `command_start`/`command_error` for every + /// rejected request went out with no installation ID at all, even on + /// an already-provisioned host -- unlike a request that makes it far + /// enough to be serviced, which always calls `refresh_ids` + /// via `servicing_request`/`reading_request`. + /// + /// `#[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: &str, + 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. + // `refresh_ids` is called 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.refresh_ids(); + 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: &str, + 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.refresh_ids(); + 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 659e88849..f62859374 100644 --- a/crates/trident/src/server/tridentserver/services/rollback.rs +++ b/crates/trident/src/server/tridentserver/services/rollback.rs @@ -60,10 +60,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( + "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( + "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 6da6c4324..d71b75d95 100644 --- a/crates/trident/src/server/tridentserver/services/streaming.rs +++ b/crates/trident/src/server/tridentserver/services/streaming.rs @@ -19,9 +19,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( + "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 b2d550c69..252aa9132 100644 --- a/crates/trident/src/server/tridentserver/services/update.rs +++ b/crates/trident/src/server/tridentserver/services/update.rs @@ -25,17 +25,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( + "update", + "stage", + "Missing staging configuration", + )); }; let Some(host_config) = staging.config else { - return Err(Status::invalid_argument( + return Err(self.reject_invalid_argument( + "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( + "update", + "finalize", + "Missing finalize configuration", + )); }; let data_store_path = self.agent_config.datastore_path().to_owned(); @@ -72,7 +82,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( + "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 0d3139650..71954d2eb 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/src/validation.rs b/crates/trident/src/validation.rs index 0f5e4242e..f9363c568 100644 --- a/crates/trident/src/validation.rs +++ b/crates/trident/src/validation.rs @@ -38,6 +38,20 @@ pub fn parse_host_config( parsed } +/// Read and parse (but do not semantically validate) a Host Configuration +/// file at the given path -- the same parse step `Trident::new` performs +/// internally before it does anything else with the file, exposed so +/// callers can perform it as a side-effect-free preflight check (e.g. +/// before creating a datastore on the strength of the path merely +/// existing) without duplicating the read+parse logic themselves. +pub fn parse_host_config_file(path: impl AsRef) -> Result { + let contents = + fs::read_to_string(path.as_ref()).structured(InvalidInputError::ReadInputFile { + path: path.as_ref().display().to_string(), + })?; + parse_host_config(&contents, Some(path.as_ref())) +} + /// Validate a Host Configuration file at the given path. pub fn validate_host_config_file(path: impl AsRef) -> Result<(), TridentError> { info!( diff --git a/crates/trident_api/src/error.rs b/crates/trident_api/src/error.rs index 1a8e4b9f3..e84fcd478 100644 --- a/crates/trident_api/src/error.rs +++ b/crates/trident_api/src/error.rs @@ -223,6 +223,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, @@ -918,6 +924,16 @@ impl TridentError { } .ok() } + + /// Returns the `file:line` location where this error was originally + /// constructed (via `TridentError::new`/`with_source`/`internal`, or + /// `ReportError::structured`), same format as the `location` field + /// already included in this type's `Serialize` impl. Useful for + /// telemetry/logging call sites that want the error's origin without + /// needing the full `Debug` context chain. + pub fn location(&self) -> String { + format!("{}:{}", self.0.location.file(), self.0.location.line()) + } } pub trait ReportError { diff --git a/docs/Reference/Telemetry.md b/docs/Reference/Telemetry.md index 0add1722e..1b4dbbacb 100644 --- a/docs/Reference/Telemetry.md +++ b/docs/Reference/Telemetry.md @@ -58,6 +58,30 @@ host along with the metrics/spans themselves: the daemon executed for a gRPC request), or `grpc-client` (the CLI acting as a client, relaying a command to a running daemon). +## Command Errors + +If a *servicing* command (`install`, `update`, `commit`, `rollback`, +`rebuild_raid`, `stream_disk`, and their gRPC/`grpc-client` equivalents) +fails, a +`command_error` event is also sent (tagged with the same +`operation_id`/`command` as above), breaking the failure down into: + +- `kind`: the top-level error category (e.g. `internal`, `invalid-input`, + `servicing`, `initialization`). +- `subkind`: the specific error within that category (e.g. + `check-root-privileges`), when one applies. +- `location`: the `file:line` in Trident's source where the error was + originally raised. + +A `grpc-client` invocation only fires its own `command_error` when the +daemon it talked to never actually responded (a transport-level failure: +the daemon's socket wasn't found, the connection was refused, or it +dropped mid-call). If the daemon did respond -- including rejecting the +request outright -- the daemon's own `command_error` for that failure +already has full `kind`/`subkind`/`location` fidelity, so `grpc-client` +stays silent rather than reporting the same failure again under a +generic classification. + ## Delivery Telemetry delivery is always best-effort and never affects servicing