diff --git a/crates/trident/Cargo.toml b/crates/trident/Cargo.toml index 3b54f2fd0..6fb6ce218 100644 --- a/crates/trident/Cargo.toml +++ b/crates/trident/Cargo.toml @@ -35,7 +35,7 @@ oci-client = { workspace = true } procfs = { workspace = true } rayon = { workspace = true } regex = { workspace = true } -reqwest = { workspace = true } +reqwest = { workspace = true, features = ["stream"] } semver = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } @@ -64,7 +64,7 @@ hyper = { workspace = true } prost-types = { workspace = true } tokio = { workspace = true } tokio-stream = { workspace = true } -tokio-util = { workspace = true } +tokio-util = { workspace = true, features = ["io-util"] } tonic = { workspace = true } tonic-middleware = { workspace = true } diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs index fdb2b3be4..03ef8a3d7 100644 --- a/crates/trident/src/datastore.rs +++ b/crates/trident/src/datastore.rs @@ -22,6 +22,14 @@ use crate::{logging::operation_context, TRIDENT_SEMVER_VERSION}; /// can be correlated. const INSTALLATION_ID_KEY: &str = "installation-id"; +/// Key under which the current servicing ID is stored in the generic +/// key-value table. Unlike `INSTALLATION_ID_KEY`, this is overwritten +/// every time a new servicing operation begins staging (install, update, +/// or manual rollback) via `DataStore::ensure_servicing_id` -- it identifies +/// "the servicing operation in progress (or last completed)", not the +/// host installation as a whole. +const SERVICING_ID_KEY: &str = "servicing-id"; + pub struct DataStore { db: Option, host_status: HostStatus, @@ -512,12 +520,11 @@ impl DataStore { /// `serde::Serialize`/`serde::de::DeserializeOwned` can be stored, not /// just `HostStatus`. /// - /// No first-party caller exists yet in this crate on its own -- this is - /// exercised only by tests for now. It's kept as part of the generic - /// key-value API (see `get_value`) for the first real consumer that - /// wants unconditional-overwrite semantics (as opposed to - /// `set_value_if_absent`'s insert-if-absent semantics). - #[allow(dead_code)] + /// `installation_id` needs insert-if-absent semantics instead (see + /// `set_value_if_absent`), but `servicing_id` -- a new servicing + /// operation genuinely does replace whatever ID (if any) came before + /// it -- is a first-party caller of this unconditional-overwrite + /// form. pub(crate) fn set_value(&self, key: &str, value: &T) -> Result<(), TridentError> { let contents = serde_json::to_string(value).structured(ServicingError::Datastore { inner: DatastoreError::SerializeValue { @@ -688,6 +695,45 @@ impl DataStore { } } + /// Returns the currently persisted servicing ID, if any. `None` if no + /// servicing operation has ever staged (via `ensure_servicing_id`) on + /// this datastore. + pub fn servicing_id(&self) -> Result, TridentError> { + self.get_value::(SERVICING_ID_KEY) + } + + /// Generates a fresh servicing ID and persists it (unconditionally + /// overwriting any previous value), returning the new ID. Called once + /// at the start of staging for install, update, or manual rollback -- + /// i.e. whenever `Operations::has_stage()` is true for that + /// invocation -- so a finalize-only call (or a later `commit`) must + /// use [`Self::servicing_id`] to read the value back instead of + /// calling this again. + /// + /// Like `ensure_installation_id`, the value used is this invocation's + /// own `operation_id` (see `logging::operation_context::current`), + /// not an unrelated freshly-generated UUID: whichever call actually + /// performs the staging is what should be correlatable, via this same + /// ID, with every later command (finalize, commit) that touches the + /// same servicing operation. Falls back to a fresh random UUID only if + /// called with no operation context active (should not happen for any + /// real caller). + /// + /// Unlike `ensure_installation_id`, this always generates a *new* ID + /// on every call -- there is no "first access wins"/get-or-create + /// semantics here, since a new servicing operation genuinely is a new + /// operation, not a value that should be stable for the datastore's + /// lifetime. (This also means, unlike installation_id, there is no + /// concurrent-first-access race to guard against: `set_value`'s plain + /// overwrite is sufficient.) + pub fn ensure_servicing_id(&mut self) -> Result { + let id = operation_context::current() + .and_then(|(operation_id, _, _)| Uuid::parse_str(&operation_id).ok()) + .unwrap_or_else(Uuid::new_v4); + self.set_value(SERVICING_ID_KEY, &id)?; + Ok(id) + } + /// Parse a single HostStatus entry from a datastore query result. /// 1. Read each row as a string containing YAML-encoded Host Status. /// 2. Decode the YAML string into a serde_yaml Value. @@ -1162,6 +1208,7 @@ mod tests { let stale_id = staging.ensure_installation_id().unwrap(); assert_ne!(stale_id, installation_id); } + let servicing_id = staging.ensure_servicing_id().unwrap(); staging.set_value("staged-image", &"image-hash").unwrap(); staging .with_host_status(|status| { @@ -1180,6 +1227,7 @@ mod tests { let mut staging = super::DataStore::open_or_create(&staging_path).unwrap(); staging.inherit_installation_id(installation_id).unwrap(); assert_eq!(staging.installation_id().unwrap(), Some(installation_id)); + assert_eq!(staging.servicing_id().unwrap(), Some(servicing_id)); assert_eq!( staging.get_value::("staged-image").unwrap(), Some("image-hash".to_string()) @@ -1194,6 +1242,9 @@ mod tests { for path in [&source_path, &staging_path, &target_path] { let datastore = super::DataStore::open(path).unwrap(); assert_eq!(datastore.installation_id().unwrap(), Some(installation_id)); + if path != &source_path { + assert_eq!(datastore.servicing_id().unwrap(), Some(servicing_id)); + } } } } @@ -1248,6 +1299,30 @@ mod tests { temp_dir.close().unwrap(); } + + #[test] + /// Regression test: `ensure_servicing_id` always generates a *fresh* ID + /// on every call, unlike `ensure_installation_id`'s get-or-create + /// semantics. This is what lets a retried clean install (e.g. a + /// previous attempt's staging failed, leaving + /// `ServicingState::NotProvisioned`) get its own distinct servicing ID + /// rather than reusing a stale one from the failed attempt. + fn test_ensure_servicing_id_generates_a_fresh_id_each_call() { + let temp_dir = tempfile::tempdir().unwrap(); + let datastore_path = temp_dir.path().join("db.sqlite"); + + let mut datastore = super::DataStore::open_or_create(&datastore_path).unwrap(); + + let first = datastore.ensure_servicing_id().unwrap(); + assert_eq!(datastore.servicing_id().unwrap(), Some(first)); + + let second = datastore.ensure_servicing_id().unwrap(); + assert_ne!( + first, second, + "ensure_servicing_id should generate a fresh ID on every call, not reuse the previous one" + ); + assert_eq!(datastore.servicing_id().unwrap(), Some(second)); + } } #[cfg(feature = "functional-test")] diff --git a/crates/trident/src/lib.rs b/crates/trident/src/lib.rs index de03e0c88..ff54965d9 100644 --- a/crates/trident/src/lib.rs +++ b/crates/trident/src/lib.rs @@ -154,7 +154,8 @@ impl Trident { /// Construct without attaching the source datastore's installation ID. /// /// The CLI's multiboot path defers attachment until `install` has persisted - /// the shared identity in the selected staging datastore. + /// the shared identity in the selected staging datastore. Startup events + /// omit persisted identities regardless of which constructor is used. pub fn new_deferring_installation_id( config_source: Option, datastore_path: &Path, @@ -245,8 +246,33 @@ impl Trident { info!("Running Trident in a container"); } - // Attach the persisted installation ID before startup metrics. Defer it - // when a multiboot install may select a different datastore. + // Attach this host's installation ID and its current servicing ID + // -- if either has already been stamped -- to the shared + // TraceStream before any startup metrics are emitted, so every + // trace/metric -- including this very "trident_start" event -- + // carries them once available. Installation ID is normally + // created by `Trident::install` (at the start of staging), so + // every other caller of `Trident::new` + // (update/commit/rollback/rebuild-raid, and every daemon RPC + // handler) just attaches whatever was already persisted at + // install time. But not purely read-only: two other paths can + // also mint one here on first attach -- + // `installation_id_or_migrate` performs a one-time migration + // write for a legacy/offline-provisioned datastore that predates + // this field (see `DataStore::installation_id_or_migrate`), and + // the CIH update-bootstrap path (below, in `update`) creates one + // via `ensure_and_attach_installation_id`. Neither ever creates a + // *datastore*, only (at most) writes into one that already + // exists. The servicing-ID half is fully read-only. + // + // Skips the installation-ID half specifically (but still attaches + // servicing ID) when `attach_installation_id` is false (see + // `new_deferring_installation_id`): attaching would stamp this + // invocation with the wrong (pre-swap) datastore's ID for a + // multiboot install that later swaps to a different datastore. + // Servicing ID is unaffected by that swap (a fresh temporary + // datastore simply has no servicing ID yet), so it's still worth + // attaching here even in that case. tracestream.attach_ids_if_present(datastore_path, !attach_installation_id); // Trace features enabled in the Host Configuration. @@ -547,6 +573,14 @@ impl Trident { warn!("Failed to attach installation ID: {e:?}"); } + // Generate (if this call has `stage` allowed, from a source + // allowed to generate persistent IDs) or read back (otherwise) + // the servicing ID for this install, and attach it. See + // `TraceStream::refresh_servicing_id`'s doc comment for the + // full rationale (best-effort, fail-closed source gate, + // deliberate unconditional regeneration on `has_stage`). + tracestream.refresh_servicing_id(datastore, allowed_operations.has_stage()); + // Use a prefetched image if provided, otherwise load the image // specified in the Host Configuration. let image = match prefetched_image { @@ -708,6 +742,18 @@ impl Trident { } } + // Generate (if this call has `stage` allowed, from a source + // allowed to generate persistent IDs) or read back (otherwise) + // the servicing ID for this update, and attach it. Placed + // here -- after the datastore is guaranteed to exist (whether + // via the CIH bootstrap above or a pre-existing persistent + // one) -- so it covers both the CIH-bootstrap and normal + // update paths uniformly, unlike installation_id (which is + // attached earlier, in `Trident::new`, for the normal path). + // See `TraceStream::refresh_servicing_id`'s doc comment for + // the full rationale. + tracestream.refresh_servicing_id(datastore, allowed_operations.has_stage()); + // The storage section is optional for updates if COSI is in use. if host_config.image.is_some() && host_config.storage == Default::default() { host_config.storage = datastore.host_status().spec.storage.clone(); @@ -861,6 +907,15 @@ impl Trident { return Ok((ExitKind::Done, ServicingType::NoActiveServicing)); } + // `commit` never generates a servicing ID -- it only ever reads + // back whatever the preceding install/update/rollback stage + // persisted, so this invocation's own (fresh) `operation_id` can + // be correlated back to that servicing operation. Always passes + // `has_stage = false`, so this only ever reads (see + // `TraceStream::refresh_servicing_id`'s doc comment). + let tracestream = self.tracestream.clone(); + tracestream.refresh_servicing_id(datastore, false); + let servicing_type = match servicing_state { ServicingState::AbUpdateFinalized | ServicingState::AbUpdateHealthCheckFailed => { ServicingType::AbUpdate @@ -962,6 +1017,16 @@ impl Trident { return Ok((ExitKind::Done, ServicingType::NoActiveServicing)); } + // Generate (if this call has `stage` allowed, from a source + // allowed to generate persistent IDs) or read back (otherwise) the + // servicing ID for this manual rollback, and attach it. Safe to do + // before `execute_and_record_error`: the datastore is already + // known-persistent per the state check above. See + // `TraceStream::refresh_servicing_id`'s doc comment for the full + // rationale. + let tracestream = self.tracestream.clone(); + tracestream.refresh_servicing_id(datastore, allowed_operations.has_stage()); + let rollback_result = self.execute_and_record_error(datastore, |datastore| { manual_rollback::execute_rollback( datastore, diff --git a/crates/trident/src/logging/operation_context.rs b/crates/trident/src/logging/operation_context.rs index 97d141747..d64e1661c 100644 --- a/crates/trident/src/logging/operation_context.rs +++ b/crates/trident/src/logging/operation_context.rs @@ -83,6 +83,27 @@ impl OperationSource { } } +/// Whether the current invocation is allowed to generate/attach a +/// persistent, datastore-backed ID (`installation_id`, `servicing_id`) -- +/// as opposed to merely reading one back for telemetry, which every +/// source may do. +/// +/// Only [`OperationSource::Cli`] and [`OperationSource::Daemon`] ever +/// actually execute `Trident::install`/`update`/`rollback`/`commit` +/// against a real datastore; [`OperationSource::GrpcClient`] only relays +/// the request to a daemon, which is what performs the real work (tagged +/// `Daemon`). There is therefore no real call site under `GrpcClient` +/// today that would try to generate one of these IDs -- but that's an +/// implicit property of today's call graph, not something the type system +/// enforces. This gate makes the assumption explicit and centrally +/// testable, rather than depending on every future call site +/// independently getting it right (an equivalent implicit assumption is +/// what caused `installation_id`'s multiboot mis-attach bug: see +/// `Trident::new_deferring_installation_id`). +pub fn should_generate_persistent_ids(source: OperationSource) -> bool { + matches!(source, OperationSource::Cli | OperationSource::Daemon) +} + thread_local! { static CURRENT_OPERATION: RefCell> = const { RefCell::new(None) }; @@ -260,6 +281,16 @@ mod tests { ); } + #[test] + fn test_should_generate_persistent_ids() { + assert!(should_generate_persistent_ids(OperationSource::Cli)); + assert!(should_generate_persistent_ids(OperationSource::Daemon)); + assert!( + !should_generate_persistent_ids(OperationSource::GrpcClient), + "grpc-client only relays to a daemon; it must never generate installation_id/servicing_id itself" + ); + } + #[test] fn test_run_with_operation_clears_context_on_panic() { assert!(current().is_none()); diff --git a/crates/trident/src/logging/tracestream.rs b/crates/trident/src/logging/tracestream.rs index edafb531d..452bb924e 100644 --- a/crates/trident/src/logging/tracestream.rs +++ b/crates/trident/src/logging/tracestream.rs @@ -1,4 +1,4 @@ -use crate::telemetry::fields; +use crate::telemetry::{events, fields}; use std::{ collections::BTreeMap, fs::File, @@ -91,6 +91,12 @@ pub struct TraceStream { // >=1.70 target: Arc>>, installation_id: Arc>>, + /// Identifies one logical servicing operation (an install, update, or + /// manual rollback) that may span multiple invocations -- a stage + /// call, a later finalize call, and a later still `commit` call each + /// get their own `operation_id`, but share this one value. See + /// `crate::datastore::DataStore::ensure_servicing_id`/`servicing_id`. + servicing_id: Arc>>, disabled: bool, } @@ -163,9 +169,54 @@ impl TraceStream { .unwrap_or(false) } - /// Attach an existing installation ID, migrating provisioned legacy - /// datastores when needed. Never creates a datastore. Multiboot - /// callers may defer attachment until their datastore is selected. + /// Best-effort attempt to attach both this host's installation ID and + /// its current servicing ID (if either is present), from a single + /// datastore connection opened once and shared between both reads -- + /// rather than opening separately per ID (the CLI dispatch, the + /// daemon's startup attach, and the daemon's per-request backstop all + /// used to call a dedicated installation-ID-only version of this, + /// each independently opening the datastore just to read one field). + /// + /// The installation-ID half already-cached from a prior call on this + /// `TraceStream`, or freshly read from the datastore at + /// `datastore_path` if one already exists there. Never creates a + /// *datastore*, and never creates an installation ID for a genuinely + /// unprovisioned host: a command that is allowed to initialize a + /// brand-new datastore (see + /// [`crate::datastore::DataStore::may_initialize_datastore_for_command`]) + /// must still call [`Self::ensure_and_attach_installation_id`] instead, + /// on a datastore handle it already owns. + /// + /// Not fully read-only, though: for a datastore that is already + /// provisioned (via offline init or the CIH update-bootstrap path, + /// both of which adopt a datastore without ever calling + /// `Trident::install`) but has no installation ID yet, this performs a + /// one-time migration *write* to mint one -- see + /// [`crate::datastore::DataStore::installation_id_or_migrate`]. Callers + /// that require true read-only behavior (e.g. a genuinely + /// unprivileged/diagnostic path) must not assume this call can never + /// write to the datastore. The servicing-ID half is fully read-only + /// (see [`Self::attach_servicing_id_if_present`]). + /// + /// `skip_installation_id` lets a caller that knows this invocation may + /// go on to swap datastores (a multiboot install) opt out of just the + /// installation-ID half -- attaching the pre-swap datastore's + /// installation ID here would misattribute this invocation's earliest + /// telemetry to the wrong host installation for the remainder of the + /// run, the same reason the installation-ID attach used to be skipped + /// entirely by `Trident::new_deferring_installation_id`'s caller + /// before this function existed. The servicing-ID half is unaffected + /// by a multiboot swap (a fresh temporary datastore simply has no + /// 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. + /// 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) { if !datastore_path.exists() { return; @@ -192,6 +243,8 @@ impl TraceStream { } } } + + self.attach_servicing_id_if_present(&datastore); } /// Ensures `datastore` has an installation ID -- creating one if this @@ -218,6 +271,146 @@ impl TraceStream { Ok(()) } + /// Set the servicing ID to attach to every trace entry sent from this + /// point forward. Expected to be called once per `Trident`/`TraceStream` + /// lifetime, either by [`Self::ensure_and_attach_servicing_id`] (the + /// call that stages a new servicing operation) or + /// [`Self::attach_servicing_id_if_present`] (a finalize-only or + /// `commit` call reading one back). + pub fn set_servicing_id(&self, servicing_id: String) { + match self.servicing_id.write() { + Ok(mut val) => { + val.replace(servicing_id); + } + Err(_) => warn!("Failed to lock tracestream to set servicing ID"), + } + } + + /// Returns a clone of the shared servicing-ID handle -- the same + /// underlying `Arc>` written by `set_servicing_id` -- so + /// other telemetry sinks (namely `AppInsightsSender`) can read the + /// current value at send-time without needing their own copy of the + /// logic that sets it. + pub fn servicing_id_handle(&self) -> Arc>> { + self.servicing_id.clone() + } + + /// Best-effort attempt to attach the currently-persisted servicing ID + /// (if any) -- read-only, never generates one. Used by a finalize-only + /// call (`Operations::has_stage() == false`) and by `commit`, both of + /// which must correlate with whatever servicing ID an earlier stage + /// call already persisted, rather than generating their own. + /// + /// Deliberately always re-reads from `datastore` rather than + /// short-circuiting on an already-cached value (unlike + /// `attach_ids_if_present`, which *is* safe to + /// cache-and-skip): `installation_id` is stable for the entire + /// lifetime of the `TraceStream`/`DataStore` it's attached to, but + /// `servicing_id` is not -- a single long-lived daemon `TraceStream` (see + /// `server::tridentserver`, which owns one `TraceStream` shared across + /// every request it serves) can legitimately see a *different* + /// already-staged servicing operation across separate finalize/commit + /// requests, e.g. one staged out-of-process via the CLI. Caching the + /// first value read here would silently keep attaching that stale ID + /// to every later finalize/commit call in the same process, even once + /// the datastore itself has moved on. + /// + /// Deliberately takes an already-open `datastore` handle rather than a + /// path (unlike `attach_ids_if_present`): by the time a + /// finalize/commit call reaches this point it already holds one, and + /// a servicing ID is only ever meaningful in the context of a + /// datastore that has already been through at least one staged + /// operation. + pub fn attach_servicing_id_if_present(&self, datastore: &DataStore) { + match datastore.servicing_id() { + Ok(Some(servicing_id)) => { + info!("Servicing ID: {servicing_id}"); + self.set_servicing_id(servicing_id.to_string()); + } + Ok(None) => { + match self.servicing_id.write() { + Ok(mut val) => { + val.take(); + } + Err(_) => warn!("Failed to lock tracestream to clear servicing ID"), + } + debug!("No servicing ID persisted yet (nothing staged on this datastore)"); + } + Err(e) => { + warn!("Failed to read servicing ID: {e:?}"); + } + } + } + + /// Generates a fresh servicing ID for a new staging operation and + /// attaches it. Called by `Trident::install`/`update`/`rollback` + /// exactly when `Operations::has_stage()` is true for that invocation + /// -- a finalize-only invocation must call + /// [`Self::attach_servicing_id_if_present`] instead, to read back the + /// value this call persists rather than generating its own. + /// + /// Best-effort: a failure here must not block the servicing operation + /// itself, matching the "telemetry never affects servicing outcomes" + /// invariant -- callers should `warn!`-and-continue on error rather + /// than propagating it with `?`. + pub fn ensure_and_attach_servicing_id( + &self, + datastore: &mut DataStore, + ) -> Result<(), TridentError> { + let servicing_id = datastore.ensure_servicing_id()?; + info!("Servicing ID: {servicing_id}"); + self.set_servicing_id(servicing_id.to_string()); + Ok(()) + } + + /// Generates (if `has_stage` is true, from a source allowed to + /// generate persistent IDs) or reads back (otherwise) the servicing ID + /// for the current invocation, and attaches it. Single shared + /// implementation for the branch used identically by + /// `Trident::install`/`update`/`rollback` (each passing their own + /// `Operations::has_stage()`) and `Trident::commit` (which always + /// passes `has_stage = false`, since `commit` never stages anything + /// itself) -- see those call sites for the full rationale. + /// + /// Reading back is unconditional and source-agnostic -- it is a + /// harmless, read-only best-effort lookup regardless of who is calling + /// -- but generating is gated on `should_generate_persistent_ids` and + /// fails closed (does not generate) if the source is unknown or not + /// allowed to generate persistent IDs (see that function's doc + /// comment). Best-effort throughout: a failure to generate is only + /// logged, matching every other persistent-ID attachment in this + /// module. + /// + /// Deliberately regenerates (overwriting any previously-staged value) + /// whenever `has_stage` is true, even if a servicing operation is + /// already staged and this call only ends up finalizing it (e.g. a + /// combined `install`/`update` re-run against an unchanged, already- + /// staged Host Configuration): any invocation that is allowed to stage + /// is entitled to a fresh servicing ID for that possibility, and + /// `has_stage` is treated as *operator intent to service* rather than + /// "staging will definitely happen." Precisely detecting the latter + /// would require the same COSI fetch (`Trident::get_cosi_image`) that + /// both the staging and finalize-only branches already need + /// unconditionally afterward (its hash is part of either branch's + /// return value) -- there's no cheap way to know the real outcome + /// before paying that cost, so this is a deliberate tradeoff, not an + /// oversight. Consequence: the original stage's servicing ID is + /// orphaned in telemetry in that finalize-only-in-practice case (no + /// single ID spans that original stage through to this invocation's + /// finalize/commit) -- acceptable, since each stage-capable invocation + /// is considered its own servicing episode by definition, regardless + /// of whether the engine ends up actually restaging anything. + pub fn refresh_servicing_id(&self, datastore: &mut DataStore, has_stage: bool) { + let source = operation_context::current().map(|(_, _, source)| source); + if has_stage && source.is_some_and(operation_context::should_generate_persistent_ids) { + if let Err(e) = self.ensure_and_attach_servicing_id(datastore) { + warn!("Failed to create servicing ID: {e:?}"); + } + } else { + self.attach_servicing_id_if_present(datastore); + } + } + /// Create a Boxed TraceSender pub fn make_trace_sender(&self) -> Box { self.make_trace_sender_with_metrics_path(TRIDENT_METRICS_FILE_PATH) @@ -233,6 +426,7 @@ impl TraceStream { Box::new(TraceSender::new( self.target.clone(), self.installation_id.clone(), + self.servicing_id.clone(), metrics_file_path, )) } @@ -241,6 +435,7 @@ impl TraceStream { pub struct TraceSender { server: Arc>>, installation_id: Arc>>, + servicing_id: Arc>>, client: reqwest::blocking::Client, metrics_file: Option, } @@ -255,11 +450,13 @@ impl TraceSender { fn new( server: Arc>>, installation_id: Arc>>, + servicing_id: Arc>>, metrics_file_path: &str, ) -> Self { Self { server, installation_id, + servicing_id, client: reqwest::blocking::Client::new(), metrics_file: match files::create_file(metrics_file_path) { Ok(f) => Some(f), @@ -277,14 +474,42 @@ impl TraceSender { self.server.read().map(|s| s.clone()).unwrap_or_default() } - fn additional_fields(&self) -> BTreeMap { + /// Build the `additional_fields` map for a trace entry: the static + /// `ADDITIONAL_FIELDS`, the installation ID (if one has been set via + /// `TraceStream::set_installation_id`), the servicing ID (if one has + /// been set via `TraceStream::set_servicing_id`), and the current + /// thread's `operation_id`/`command` (if any, see `operation_context`), + /// so entries can be correlated back to a specific host installation + /// and servicing operation. + /// + /// Startup events omit both persisted IDs. Other events retain the + /// cached IDs, with a staging command's servicing ID overridden by + /// its operation ID in `merge_operation_context`. + /// + /// `operation_id`/`command` are deliberately merged here rather than + /// into the metric's own `value` (as scalar/span fields are): mixing + /// them into `value` would change the established schema for simple + /// scalar metrics -- e.g. `clean_install_start` would go from + /// `"value": true` to `"value": {"command": ..., "operation_id": ..., + /// "value": true}` the moment it ran inside an operation context, + /// breaking that contract for existing consumers. + /// + /// `installation_id` is attached only after reading or creating it in + /// the selected datastore. Until then it stays absent: an unloaded ID + /// does not establish that this command is creating a new installation. + fn additional_fields(&self, metric_name: &str) -> BTreeMap { let mut fields = ADDITIONAL_FIELDS.clone(); if let Ok(installation_id) = self.installation_id.read() { if let Some(installation_id) = installation_id.as_ref() { fields.insert(fields::INSTALLATION_ID.to_string(), json!(installation_id)); } } - merge_operation_context(&mut fields); + if let Ok(servicing_id) = self.servicing_id.read() { + if let Some(servicing_id) = servicing_id.as_ref() { + fields.insert(fields::SERVICING_ID.to_string(), json!(servicing_id)); + } + } + merge_operation_context(metric_name, &mut fields); fields } @@ -353,11 +578,12 @@ where .unwrap_or_default() }; + let additional_fields = self.additional_fields(&metric_name); let entry = TraceEntry { timestamp: Utc::now(), metric_name, value: json!(value), - additional_fields: self.additional_fields(), + additional_fields, platform_info: PLATFORM_INFO.clone(), }; @@ -437,7 +663,7 @@ where timestamp: Utc::now(), metric_name: span.name().to_string(), value: json!(visitor.fields), - additional_fields: self.additional_fields(), + additional_fields: self.additional_fields(span.name()), platform_info: PLATFORM_INFO.clone(), }; @@ -497,7 +723,8 @@ fn product_uuid_or_unknown(uuid: Result) -> String { /// Merge the current thread's `operation_id`/`command`/`source` (see /// `operation_context`), if any, into `fields`. Values the caller already /// set (e.g. an event that explicitly names its own `command`) are never -/// overwritten. +/// overwritten, except for staging servicing IDs. Both persisted identity +/// fields are always removed from `command_start` and `trident_start`. /// /// Shared by both local telemetry sinks (`TraceSender::additional_fields` /// below and `AppInsightsSender::send_event`) so the @@ -506,7 +733,7 @@ fn product_uuid_or_unknown(uuid: Result) -> String { /// prior version of this function existed independently in each sink, /// which risked the two silently diverging if the rule ever changed in /// only one place. -pub(crate) fn merge_operation_context(fields: &mut BTreeMap) { +pub(crate) fn merge_operation_context(metric_name: &str, fields: &mut BTreeMap) { if let Some((operation_id, command, source)) = operation_context::current() { fields .entry(fields::OPERATION_ID.to_string()) @@ -517,6 +744,26 @@ pub(crate) fn merge_operation_context(fields: &mut BTreeMap) { fields .entry(fields::SOURCE.to_string()) .or_insert_with(|| json!(source.as_str())); + // A cached servicing_id can go genuinely + // stale mid-process: a long-lived daemon's cached + // `TraceStream::servicing_id` still holds a *previous* operation's + // value until this invocation's own + // `ensure_and_attach_servicing_id` runs deeper in the call stack + // and overwrites it. For the commands that will end up + // generating their own new servicing ID anyway (see + // `CommandKind::generates_servicing_id`'s doc comment for the + // exact set and why), that value is already fully determined the + // moment this operation's context exists -- so force it here + // (unconditionally, not just filling a missing value) rather than + // leaving a servicing event showing a stale, unrelated + // operation's ID. + if command.generates_servicing_id() { + fields.insert(fields::SERVICING_ID.to_string(), json!(operation_id)); + } + } + if matches!(metric_name, events::COMMAND_START | events::TRIDENT_START) { + fields.remove(fields::INSTALLATION_ID); + fields.remove(fields::SERVICING_ID); } } @@ -591,12 +838,12 @@ mod tests { operation_context::OperationSource::Cli, || { let mut fields = BTreeMap::new(); - merge_operation_context(&mut fields); + merge_operation_context("test_metric", &mut fields); assert!(fields.contains_key("operation_id")); assert!(!fields.contains_key("installation_id")); fields.insert("installation_id".to_string(), json!("persisted-id")); - merge_operation_context(&mut fields); + merge_operation_context("test_metric", &mut fields); assert_eq!(fields["installation_id"], json!("persisted-id")); }, ); @@ -708,6 +955,82 @@ mod tests { ); } + #[test] + fn test_startup_events_omit_ids_without_operation_context() { + for name in ["command_start", "trident_start"] { + let mut fields = BTreeMap::from([ + ("installation_id".to_string(), json!("installation")), + ("servicing_id".to_string(), json!("servicing")), + ]); + merge_operation_context(name, &mut fields); + assert!(!fields.contains_key("installation_id")); + assert!(!fields.contains_key("servicing_id")); + assert!(!fields.contains_key("operation_id")); + } + } + + #[test] + fn test_startup_events_omit_persisted_ids() { + for command in [ + CommandKind::install_stage(), + CommandKind::install_finalize(), + ] { + for source in [ + operation_context::OperationSource::Cli, + operation_context::OperationSource::Daemon, + operation_context::OperationSource::GrpcClient, + ] { + let directory = tempfile::tempdir().unwrap(); + let metrics_path = directory.path().join("metrics.jsonl"); + let stream = TraceStream::default(); + stream.set_installation_id("shared-installation".to_string()); + stream.set_servicing_id("previous-servicing".to_string()); + let sender = stream + .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap()) + .with_filter(filter::LevelFilter::INFO); + let _guard = tracing::subscriber::set_default( + tracing_subscriber::Registry::default().with(sender), + ); + + let operation_id = operation_context::run_with_operation(&command, source, || { + tracing::info!(metric_name = "trident_start", value = 42.0); + tracing::info!(metric_name = "clean_install_start", value = true); + operation_context::current().unwrap().0 + }); + let entries: Vec = BufReader::new(File::open(&metrics_path).unwrap()) + .lines() + .map(|line| serde_json::from_str(&line.unwrap()).unwrap()) + .collect(); + assert_eq!(entries.len(), 3); + for (entry, name) in + entries + .iter() + .zip(["command_start", "trident_start", "clean_install_start"]) + { + assert_eq!(entry["metric_name"], name); + let fields = &entry["additional_fields"]; + assert_eq!(fields["operation_id"], operation_id); + assert_eq!(fields["command"], command.as_str()); + assert_eq!(fields["source"], source.as_str()); + if name != "clean_install_start" { + assert!(fields.get("installation_id").is_none()); + assert!(fields.get("servicing_id").is_none()); + } else { + assert_eq!(fields["installation_id"], "shared-installation"); + let servicing_id = if command.generates_servicing_id() { + operation_id.as_str() + } else { + "previous-servicing" + }; + assert_eq!(fields["servicing_id"], servicing_id); + } + } + assert_eq!(entries[1]["value"], 42.0); + assert_eq!(entries[2]["value"], true); + } + } + } + #[test] /// Regression test: `TraceStream::set_installation_id` must actually /// reach the serialized trace entry's `additional_fields.installation_id` @@ -752,6 +1075,105 @@ mod tests { ); } + #[test] + /// Regression test for the daemon stale-servicing_id bug: a + /// long-lived process's cached `servicing_id` (left over from a prior + /// operation) must NOT leak onto a later staging command's own + /// servicing events. `merge_operation_context` + /// must force `servicing_id` to this invocation's own `operation_id` + /// for any command where `CommandKind::generates_servicing_id` is true, overriding + /// whatever is currently cached rather than merely filling a gap. + fn test_tracestream_servicing_id_forced_to_operation_id_for_staging_commands() { + let temp_dir = tempfile::tempdir().unwrap(); + let metrics_path = temp_dir.path().join("metrics.jsonl"); + let tracestream = TraceStream::default(); + // Simulate a stale value left over from a previous operation in + // the same long-lived process. + tracestream.set_servicing_id("stale-previous-operation-id".to_string()); + let trace_sender = tracestream + .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap()) + .with_filter(filter::LevelFilter::INFO); + + let _guard = tracing::subscriber::set_default( + tracing_subscriber::Registry::default().with(trace_sender), + ); + + let operation_id = operation_context::run_with_operation( + &CommandKind::install_stage(), + operation_context::OperationSource::Cli, + || { + let operation_id = operation_context::current() + .expect("operation context should be installed") + .0; + tracing::info!(metric_name = "test_metric_during_staging_command"); + operation_id + }, + ); + + 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(); + + let expected = format!(r#""servicing_id":"{operation_id}""#); + let metric_found = lines.iter().any(|line| { + line.contains(r#""metric_name":"test_metric_during_staging_command""#) + && line.contains(&expected) + && !line.contains("stale-previous-operation-id") + }); + + assert!( + metric_found, + "Expected staging command's metric to carry this operation's own \ + operation_id as servicing_id, not the stale cached value" + ); + } + + #[test] + /// Negative counterpart: a non-staging command (e.g. `commit`, which + /// only ever reads back a previous operation's servicing_id) must NOT + /// have its cached servicing_id overridden -- the whole point of + /// read-back is to surface exactly that previously-staged value. + fn test_tracestream_servicing_id_not_forced_for_non_staging_commands() { + let temp_dir = tempfile::tempdir().unwrap(); + let metrics_path = temp_dir.path().join("metrics.jsonl"); + let tracestream = TraceStream::default(); + tracestream.set_servicing_id("previously-staged-id".to_string()); + let trace_sender = tracestream + .make_trace_sender_with_metrics_path(metrics_path.to_str().unwrap()) + .with_filter(filter::LevelFilter::INFO); + + let _guard = tracing::subscriber::set_default( + tracing_subscriber::Registry::default().with(trace_sender), + ); + + operation_context::run_with_operation( + &CommandKind::commit(), + operation_context::OperationSource::Cli, + || { + tracing::info!(metric_name = "test_metric_during_commit"); + }, + ); + + 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(); + + let metric_found = lines.iter().any(|line| { + line.contains(r#""metric_name":"test_metric_during_commit""#) + && line.contains(r#""servicing_id":"previously-staged-id""#) + }); + + assert!( + metric_found, + "Expected commit's metric to keep the previously-staged servicing_id, \ + not have it overridden" + ); + } + #[test] fn test_tracestream_write_span_metric_to_file() { let temp_dir = tempfile::tempdir().unwrap(); @@ -792,6 +1214,79 @@ mod tests { #[tracing::instrument(name = "test_span", skip_all)] fn simulate_function_span() {} + #[test] + /// Regression test closing the stage/read-back join gap: a servicing + /// ID generated by one `TraceStream` (standing in for the process that + /// staged an install/update/rollback) must be recoverable by a + /// completely separate, unrelated `TraceStream` instance that only + /// shares the same on-disk datastore (standing in for a later + /// finalize/commit request, or a different daemon request in the same + /// process). Every other servicing_id test either exercises the + /// datastore layer alone (`datastore.rs`) or injects a pre-built + /// handle directly (`appinsights.rs`'s functional test) -- neither + /// proves the real `refresh_servicing_id` -> `DataStore::servicing_id` + /// join actually works end-to-end. + fn test_servicing_id_stage_then_read_back_across_separate_tracestreams() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = temp_dir.path().join("db.sqlite"); + + // Stage: a fresh TraceStream + datastore, as if this were the + // process running `Trident::install`'s staging half. + let mut staging_datastore = DataStore::open_or_create(&db_path).unwrap(); + let staging_tracestream = TraceStream::default(); + operation_context::run_with_operation( + &CommandKind::install(), + operation_context::OperationSource::Cli, + || { + staging_tracestream.refresh_servicing_id(&mut staging_datastore, true); + }, + ); + let staged_id = staging_tracestream + .servicing_id_handle() + .read() + .unwrap() + .clone() + .expect("staging call should have generated a servicing ID"); + + // Read back: a completely separate TraceStream + DataStore handle + // opened against the same file, as if this were a later + // finalize/commit request (or a different daemon request in the + // same long-lived process). + let mut readback_datastore = DataStore::open_or_create(&db_path).unwrap(); + let readback_tracestream = TraceStream::default(); + readback_tracestream.refresh_servicing_id(&mut readback_datastore, false); + + assert_eq!( + readback_tracestream + .servicing_id_handle() + .read() + .unwrap() + .clone(), + Some(staged_id), + "a separate TraceStream reading the same datastore should recover the staged servicing ID" + ); + } + + #[test] + /// Negative counterpart to the join test above: a `TraceStream` that + /// only ever reads (never stages) against a datastore that has never + /// had anything staged must leave the servicing ID handle `None`, + /// rather than fabricating one. + fn test_servicing_id_absent_when_never_staged() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = temp_dir.path().join("db.sqlite"); + + let mut datastore = DataStore::open_or_create(&db_path).unwrap(); + let tracestream = TraceStream::default(); + tracestream.refresh_servicing_id(&mut datastore, false); + + assert_eq!( + tracestream.servicing_id_handle().read().unwrap().clone(), + None, + "servicing ID must stay absent when nothing has ever staged on this datastore" + ); + } + #[test] /// Same cross-invocation join gap as /// `test_servicing_id_stage_then_read_back_across_separate_tracestreams`, @@ -869,6 +1364,20 @@ mod tests { "installation ID must stay absent when none has ever been created on this datastore" ); } + + #[test] + fn test_idless_datastore_clears_previously_cached_servicing_id() { + let temp_dir = tempfile::tempdir().unwrap(); + let mut datastore = DataStore::open_or_create(&temp_dir.path().join("db.sqlite")).unwrap(); + let tracestream = TraceStream::default(); + let shared_handle = tracestream.servicing_id_handle(); + tracestream.set_servicing_id(uuid::Uuid::new_v4().to_string()); + + tracestream.refresh_servicing_id(&mut datastore, false); + + assert!(shared_handle.read().unwrap().is_none()); + assert!(datastore.servicing_id().unwrap().is_none()); + } } #[cfg(feature = "functional-test")] diff --git a/crates/trident/src/main.rs b/crates/trident/src/main.rs index 41ccace3f..399491933 100644 --- a/crates/trident/src/main.rs +++ b/crates/trident/src/main.rs @@ -186,25 +186,11 @@ fn run_trident( } ); - // Attach this host's installation ID and current servicing - // ID to the shared TraceStream before run_with_operation - // below fires command_start: Trident::new (further down, - // inside the closure) is the usual place installation ID - // gets attached, but that's too late for command_start, - // which run_with_operation fires immediately, before the - // closure even runs. Does not create a *datastore* (see - // `TraceStream::attach_ids_if_present`) -- silently does - // nothing if the datastore doesn't exist yet, which is - // expected for a host's first-ever install. But not purely - // read-only: on an existing datastore that predates - // `installation_id`, this can perform a one-time migration - // write to mint one (see - // `DataStore::installation_id_or_migrate`). Skips the - // installation-ID half specifically (but still attaches - // servicing ID, unaffected by a multiboot swap) when - // `defer_installation_id` is set, for the same reason - // `Trident::new_deferring_installation_id` is used below. - // + // Pre-warm persisted IDs for errors that occur before Trident::new can + // attach them. Startup events omit both IDs even when they are cached. + // This does not create a datastore, but may migrate a legacy installation + // ID. Multiboot defers installation-ID attachment until its staging + // datastore inherits the source identity. // 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 diff --git a/crates/trident/src/telemetry.rs b/crates/trident/src/telemetry.rs index 095448267..15f8b7328 100644 --- a/crates/trident/src/telemetry.rs +++ b/crates/trident/src/telemetry.rs @@ -22,6 +22,7 @@ pub(crate) mod fields { pub const METRIC_NAME: &str = "metric_name"; pub const OPERATION_ID: &str = "operation_id"; pub const OS_RELEASE: &str = "os_release"; + pub const SERVICING_ID: &str = "servicing_id"; pub const SOURCE: &str = "source"; pub const TOTAL_CPU: &str = "total_cpu"; pub const TOTAL_MEMORY_GIB: &str = "total_memory_gib";