diff --git a/.pipelines/templates/stages/trident_rpms/release.yml b/.pipelines/templates/stages/trident_rpms/release.yml index 0cd2cd86a..fb0bbe6ba 100644 --- a/.pipelines/templates/stages/trident_rpms/release.yml +++ b/.pipelines/templates/stages/trident_rpms/release.yml @@ -58,6 +58,10 @@ steps: version=$(echo $full_version | cut -d'-' -f1) prerelease=$(echo $full_version | cut -d'-' -f2-) + # Application Insights connection string identifying best-effort + # telemetry as coming from Trident's own CI/CD pipeline builds + AZURE_MONITOR_CONNECTION_STRING="InstrumentationKey=e32fc20f-2cc6-4d86-9e12-ab5d24b366f7;IngestionEndpoint=https://eastus2-3.in.applicationinsights.azure.com/;LiveEndpoint=https://eastus2.livediagnostics.monitor.azure.com/;ApplicationId=fb8e8afb-99bd-4143-ae31-22f9070005e2" + # Build RPMs and export only the artifact tarball (no image load/unpack). # CARGO_REGISTRIES_BMP_PUBLICPACKAGES_TOKEN is populated by the CargoAuthenticate task. outdir="/tmp/_rpm_artifacts" @@ -80,6 +84,7 @@ steps: --build-arg RPM_PACKAGES="$RPM_PACKAGES" \ --build-arg RUST_PACKAGE="$RUST_PACKAGE" \ --build-arg RPM_DEST="$RPM_DEST" \ + --build-arg AZURE_MONITOR_CONNECTION_STRING="$AZURE_MONITOR_CONNECTION_STRING" \ --target artifact \ --output type=local,dest="$outdir" \ . diff --git a/crates/trident/build.rs b/crates/trident/build.rs index 3e838be70..2d92b7b41 100644 --- a/crates/trident/build.rs +++ b/crates/trident/build.rs @@ -1,4 +1,5 @@ fn main() -> Result<(), Box> { println!("cargo:rerun-if-env-changed=TRIDENT_VERSION"); + println!("cargo:rerun-if-env-changed=AZURE_MONITOR_CONNECTION_STRING"); Ok(()) -} +} diff --git a/crates/trident/src/agentconfig.rs b/crates/trident/src/agentconfig.rs index 4c37a330e..440b6ca1e 100644 --- a/crates/trident/src/agentconfig.rs +++ b/crates/trident/src/agentconfig.rs @@ -7,30 +7,65 @@ use trident_api::{ error::TridentError, }; +/// Whether Trident should attempt to send tracing data to Application +/// Insights (best-effort, and only when a connection string was compiled +/// into the binary -- see [`crate::AZURE_MONITOR_CONNECTION_STRING`]). +/// +/// Defaults to [`TelemetryPreference::OptOut`]: telemetry is disabled unless +/// a user has explicitly opted in via the Agent Configuration file. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum TelemetryPreference { + /// Telemetry is disabled. Trident will not send any tracing data off + /// the host. + #[default] + OptOut, + /// Telemetry is enabled, best-effort, provided a connection string was + /// compiled into this Trident binary. + OptIn, +} + pub struct AgentConfig { datastore: PathBuf, + telemetry: TelemetryPreference, } impl AgentConfig { /// Load the AgentConfig from the default configuration file. pub fn load() -> Result { + Self::load_from_path(AGENT_CONFIG_PATH) + } + + /// Load the AgentConfig from an arbitrary path. Split out from [`load`] + /// so the parsing logic can be unit tested without touching + /// [`AGENT_CONFIG_PATH`]. + fn load_from_path(path: &str) -> Result { let mut config = Self { datastore: TRIDENT_DATASTORE_PATH_DEFAULT.into(), + telemetry: TelemetryPreference::default(), }; - if let Ok(contents) = std::fs::read_to_string(AGENT_CONFIG_PATH) { + if let Ok(contents) = std::fs::read_to_string(path) { for line in contents.lines() { - if let Some(path) = line.strip_prefix("DatastorePath=") { - config.datastore = path.trim().into(); + if let Some(value) = line.strip_prefix("DatastorePath=") { + config.datastore = value.trim().into(); + } else if let Some(value) = line.strip_prefix("Telemetry=") { + config.telemetry = match value.trim().to_ascii_lowercase().as_str() { + "optin" => TelemetryPreference::OptIn, + "optout" => TelemetryPreference::OptOut, + other => { + debug!( + "Unrecognized Telemetry setting '{other}' in agent \ + configuration file, defaulting to OptOut" + ); + TelemetryPreference::OptOut + } + }; } } } else { // If the config file does not exist, we proceed with defaults. // Only log this at debug level to avoid alarming users unnecessarily. - debug!( - "Agent configuration file not found at {}, using defaults", - AGENT_CONFIG_PATH - ); + debug!("Agent configuration file not found at {path}, using defaults"); } Ok(config) @@ -40,4 +75,85 @@ impl AgentConfig { pub fn datastore_path(&self) -> &Path { &self.datastore } + + /// Whether telemetry (best-effort tracing to Application Insights) is + /// enabled per the agent configuration file. Defaults to `false` + /// (opt-out) when unset or unrecognized. + pub fn telemetry_enabled(&self) -> bool { + matches!(self.telemetry, TelemetryPreference::OptIn) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_defaults_when_file_missing() { + let config = AgentConfig::load_from_path("/nonexistent/path/for/trident-tests.conf") + .expect("load_from_path should not fail even if the file is missing"); + assert_eq!( + config.datastore_path(), + Path::new(TRIDENT_DATASTORE_PATH_DEFAULT) + ); + assert!( + !config.telemetry_enabled(), + "telemetry must default to OptOut" + ); + } + + #[test] + fn test_telemetry_optin() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("trident.conf"); + std::fs::write(&path, "Telemetry=OptIn\n").unwrap(); + + let config = AgentConfig::load_from_path(path.to_str().unwrap()).unwrap(); + assert!(config.telemetry_enabled()); + } + + #[test] + fn test_telemetry_optout_explicit() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("trident.conf"); + std::fs::write(&path, "Telemetry=OptOut\n").unwrap(); + + let config = AgentConfig::load_from_path(path.to_str().unwrap()).unwrap(); + assert!(!config.telemetry_enabled()); + } + + #[test] + fn test_telemetry_is_case_insensitive() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("trident.conf"); + std::fs::write(&path, "Telemetry=OPTIN\n").unwrap(); + + let config = AgentConfig::load_from_path(path.to_str().unwrap()).unwrap(); + assert!(config.telemetry_enabled()); + } + + #[test] + fn test_telemetry_unrecognized_value_defaults_optout() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("trident.conf"); + std::fs::write(&path, "Telemetry=maybe\n").unwrap(); + + let config = AgentConfig::load_from_path(path.to_str().unwrap()).unwrap(); + assert!(!config.telemetry_enabled()); + } + + #[test] + fn test_datastore_and_telemetry_together() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("trident.conf"); + std::fs::write( + &path, + "DatastorePath=/custom/path.sqlite\nTelemetry=OptIn\n", + ) + .unwrap(); + + let config = AgentConfig::load_from_path(path.to_str().unwrap()).unwrap(); + assert!(config.telemetry_enabled()); + assert_eq!(config.datastore_path(), Path::new("/custom/path.sqlite")); + } } diff --git a/crates/trident/src/datastore.rs b/crates/trident/src/datastore.rs index 712a6b9a0..950bf2a03 100644 --- a/crates/trident/src/datastore.rs +++ b/crates/trident/src/datastore.rs @@ -9,18 +9,36 @@ use trident_api::{ error::{ DatastoreError, InternalError, ReportError, ServicingError, TridentError, TridentResultExt, }, - status::{decode_host_status, HostStatus, TridentVersion}, + status::{decode_host_status, HostStatus, ServicingState, TridentVersion}, }; -use crate::TRIDENT_SEMVER_VERSION; +use crate::{logging::operation_context, TRIDENT_SEMVER_VERSION}; -/// Key under which the datastore's unique database ID is stored in the -/// generic key-value table. This ID is generated once (on first access) and -/// persisted for the lifetime of the datastore. It is intended to be added to -/// tracing/telemetry so that all activity for a given host installation can +/// Key under which the datastore's stable database ID is stored in the +/// generic key-value table. This ID is generated once, on first access +/// (see `DataStore::datastore_id`), and persisted for the entire lifetime +/// of the datastore -- unlike `INSTALLATION_ID_KEY`, it is not tied to a +/// specific `Trident::install` invocation. It is intended to be added to +/// tracing/telemetry so that all activity against a given datastore can /// be correlated. const DATASTORE_ID_KEY: &str = "datastore-id"; +/// Key under which the datastore's unique installation ID is stored in the +/// generic key-value table. This ID is generated once, at the start of +/// `Trident::install` (see `DataStore::create_installation_id`), and +/// persisted for the lifetime of the datastore. It is intended to be added +/// to tracing/telemetry so that all activity for a given host installation +/// 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::new_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, @@ -220,7 +238,7 @@ impl DataStore { TridentVersion::SemVer(TRIDENT_SEMVER_VERSION.clone()); Self::write_host_status(&persistent_db, self.host_status())?; - // Carry over any generic key-value entries (e.g. the database ID) + // Carry over any generic key-value entries (e.g. the installation ID) // recorded in the temporary datastore into the persistent one, so // they survive the transition from temporary to persistent // storage. @@ -447,13 +465,11 @@ impl DataStore { /// `serde::Serialize`/`serde::de::DeserializeOwned` can be stored, not /// just `HostStatus`. /// - /// `datastore_id` is currently the only first-party caller of the - /// generic key-value store, and it needs insert-if-absent semantics - /// (see `set_value_if_absent`) rather than an unconditional overwrite, - /// so this unconditional-overwrite variant is presently exercised only - /// by tests. It's kept as part of the generic key-value API (see - /// `get_value`) for future callers that do want overwrite 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(InternalError::SerializeValue { key: key.to_string(), @@ -469,9 +485,9 @@ impl DataStore { /// Like [`Self::set_value`], but only inserts a row if `key` does not /// already have one; an existing row is left untouched. Used where two /// datastore connections could race to perform "first access" - /// initialization of a key (see [`Self::datastore_id`]): whichever - /// connection's insert commits first wins, and the other's insert - /// becomes a no-op instead of overwriting the winner's value. + /// initialization of a key (see [`Self::create_installation_id`]): + /// whichever connection's insert commits first wins, and the other's + /// insert becomes a no-op instead of overwriting the winner's value. pub(crate) fn set_value_if_absent( &self, key: &str, @@ -530,7 +546,40 @@ impl DataStore { Ok(()) } - /// Retrieve this datastore's unique database ID, generating and + /// Retrieve this datastore's stable database ID, generating and + /// persisting a new one on first access. Stable for the lifetime of + /// the datastore (survives the temporary-to-persistent transition + /// performed by `persist`). Unlike `installation_id`, this is not + /// tied to any specific `Trident::install` invocation -- it identifies + /// the datastore itself, not a servicing operation. + /// + /// Uses `set_value_if_absent` rather than an unconditional overwrite + /// so that two datastore connections racing to perform "first access" + /// initialization can't clobber each other's value: whichever + /// connection's insert commits first wins, and the loser reads back + /// whichever ID actually won that race. + pub fn datastore_id(&mut self) -> Result { + if let Some(id) = self.get_value::(DATASTORE_ID_KEY)? { + return Ok(id); + } + + self.set_value_if_absent(DATASTORE_ID_KEY, &Uuid::new_v4())?; + + self.get_value::(DATASTORE_ID_KEY)? + .structured(InternalError::Internal( + "Datastore ID missing immediately after being inserted", + )) + } + + /// Returns this datastore's installation ID, if one has already been + /// persisted. Read-only: never generates or persists one -- callers + /// that need get-or-create semantics must call + /// [`Self::create_installation_id`] instead. + pub fn installation_id(&self) -> Result, TridentError> { + self.get_value::(INSTALLATION_ID_KEY) + } + + /// Retrieve this datastore's unique installation ID, generating and /// persisting a new one on first access. /// /// This ID is stable for the lifetime of the datastore (surviving the @@ -538,6 +587,18 @@ impl DataStore { /// intended to be attached to tracing/telemetry so that activity for a /// given host installation can be correlated across logs and traces. /// + /// The value used on first access is this invocation's own + /// `operation_id` (see `logging::operation_context::current`), not an + /// unrelated freshly-generated UUID: whichever command actually + /// creates the datastore (normally `Trident::install`) has, by the + /// time it gets here, already been tagging its own telemetry with + /// `operation_id` since `command_start`. Reusing it here means that + /// invocation's telemetry never needs a second, disconnected + /// installation ID -- and every later command on this host just reads + /// back this same value. Falls back to a fresh random UUID only if + /// called with no operation context active (should not happen for any + /// real caller). + /// /// First access is not a simple read-then-write: `Trident::new` may be /// invoked concurrently (e.g. by multiple daemon RPC handlers), each /// opening its own connection to the same datastore file. A naive @@ -548,19 +609,96 @@ impl DataStore { /// freshly generated ID with `ON CONFLICT DO NOTHING` (a no-op if /// another connection already inserted one first), then read back /// whichever ID actually won that race. - pub fn datastore_id(&mut self) -> Result { - if let Some(id) = self.get_value::(DATASTORE_ID_KEY)? { + pub fn create_installation_id(&mut self) -> Result { + if let Some(id) = self.get_value::(INSTALLATION_ID_KEY)? { return Ok(id); } - self.set_value_if_absent(DATASTORE_ID_KEY, &Uuid::new_v4())?; + let new_id = operation_context::current() + .and_then(|(operation_id, _, _)| Uuid::parse_str(&operation_id).ok()) + .unwrap_or_else(Uuid::new_v4); + self.set_value_if_absent(INSTALLATION_ID_KEY, &new_id)?; - self.get_value::(DATASTORE_ID_KEY)? + self.get_value::(INSTALLATION_ID_KEY)? .structured(InternalError::Internal( - "Datastore ID missing immediately after being inserted", + "Installation ID missing immediately after being inserted", )) } + /// Returns this datastore's installation ID, creating one as a + /// one-time migration if it is missing. Single source of truth for + /// every caller that attaches an installation ID to the shared + /// `TraceStream` on datastore open, so the same migration logic isn't + /// hand-copied (and drifting) across `Trident::new`, the CLI's + /// installation-ID pre-warm, and the daemon's per-request backstop. + /// + /// A datastore can be genuinely provisioned (`servicing_state != + /// NotProvisioned`) without ever having gone through + /// `Trident::install` -- offline initialization and the CIH update + /// bootstrap both create/adopt a datastore directly. Those hosts + /// would otherwise never get an installation ID, since nothing else + /// ever calls `create_installation_id`. Create one as a one-time + /// migration in that specific case, while leaving a genuinely + /// unprovisioned (temporary or not-yet-installed) datastore alone -- + /// callers should not observe an installation ID appear before the + /// host actually completes an install. + pub fn installation_id_or_migrate(&mut self) -> Result, TridentError> { + match self.installation_id()? { + Some(id) => Ok(Some(id)), + None if self.host_status().servicing_state != ServicingState::NotProvisioned => { + self.create_installation_id().map(Some) + } + None => Ok(None), + } + } + + /// Single source of truth for "may this command/request proceed + /// without an existing datastore, because it can legitimately stage a + /// brand-new install/update itself?", shared by the CLI's dispatch + /// (`main.rs`) and the daemon's request handling + /// (`server/tridentserver/mod.rs`), so both surfaces answer this + /// question identically instead of each re-deriving it independently. + /// + /// `name` is the same stage/finalize-aware command/request name used + /// for `command`/`operation_id` telemetry (see `command_name` in + /// `main.rs`, and the literal names gRPC services pass to + /// `servicing_request`). Only names that can actually *stage* a new + /// install/update (`install`, `install_stage`, `update`, + /// `update_stage`), plus `stream_disk` (a direct-streaming install + /// path that also legitimately creates a datastore from nothing), may + /// proceed without one. Finalize-only names (`install_finalize`, + /// `update_finalize`) cannot stage anything themselves -- they require + /// an existing staged state, so letting them proceed without a + /// datastore would accept a request that has no staged operation to + /// finalize. + pub fn may_initialize_datastore_for_command(name: &str) -> bool { + matches!( + name, + "install" | "install_stage" | "update" | "update_stage" | "stream_disk" + ) + } + + /// Returns the currently persisted servicing ID, if any. `None` if no + /// servicing operation has ever staged (via `new_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. + /// + /// Unlike `create_installation_id`, this always generates a *new* ID -- + /// there is no "first access wins" semantics here, since a new + /// servicing operation genuinely is a new operation, not a value that + /// should be stable for the datastore's lifetime. + pub fn new_servicing_id(&mut self) -> Result { + let id = 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. @@ -597,6 +735,55 @@ impl DataStore { #[cfg(test)] mod tests { + #[test] + /// `may_initialize_datastore_for_command` is the single shared + /// classifier answering "may this command/request proceed without an + /// existing datastore, because it can legitimately stage a brand-new + /// install/update itself?" for both the CLI (`main.rs`) and the daemon + /// (`server/tridentserver/mod.rs`). Stage names (and `stream_disk`) + /// may; finalize-only names must not, since a finalize-only request + /// cannot itself stage anything -- it requires an existing staged + /// state. Also covers the `_noop` command-name variants + /// (`command_name` in `main.rs` emits these when neither stage nor + /// finalize is requested): they must stay denied even though they + /// aren't explicitly excluded by name, so a future refactor that + /// widens the allow-list can't silently start accepting requests with + /// no datastore for no-op invocations without this test catching it. + fn test_may_initialize_datastore_for_command() { + for name in [ + "install", + "install_stage", + "update", + "update_stage", + "stream_disk", + ] { + assert!( + super::DataStore::may_initialize_datastore_for_command(name), + "{name} should be allowed to initialize a datastore" + ); + } + for name in [ + "install_finalize", + "update_finalize", + "rollback", + "rollback_stage", + "rollback_finalize", + "commit", + "check_root", + "rebuild_raid", + "install_noop", + "update_noop", + "", + "installer", + "updated", + ] { + assert!( + !super::DataStore::may_initialize_datastore_for_command(name), + "{name} should NOT be allowed to initialize a datastore" + ); + } + } + #[test] fn test_make_datastore() { let temp_dir = tempfile::tempdir().unwrap(); @@ -631,15 +818,15 @@ mod tests { let datastore_path = temp_dir.path().join("db.sqlite"); let mut datastore = super::DataStore::open_or_create(&datastore_path).unwrap(); - let datastore_id = datastore.datastore_id().unwrap(); + let installation_id = datastore.create_installation_id().unwrap(); // Persist to the exact same path the datastore is currently open at. datastore.persist(&datastore_path).unwrap(); assert_eq!( - datastore.datastore_id().unwrap(), - datastore_id, - "Datastore ID should survive a self-persist" + datastore.create_installation_id().unwrap(), + installation_id, + "Installation ID should survive a self-persist" ); } @@ -712,7 +899,7 @@ mod tests { #[test] /// Regression test: a datastore created by an older Trident version that /// predates the `keyvalue` table (only `hoststatus` exists) must still - /// be usable after `open()` -- in particular, `datastore_id()` must + /// be usable after `open()` -- in particular, `create_installation_id()` must /// not fail with "no such table: keyvalue". fn test_open_upgrades_pre_existing_datastore_schema() { let temp_dir = tempfile::tempdir().unwrap(); @@ -734,7 +921,7 @@ mod tests { let mut datastore = super::DataStore::open(&path).unwrap(); // Should not fail with "no such table: keyvalue". - datastore.datastore_id().unwrap(); + datastore.create_installation_id().unwrap(); temp_dir.close().unwrap(); } @@ -761,14 +948,14 @@ mod tests { } #[test] - fn test_datastore_id_concurrent_first_access_is_consistent() { + fn test_installation_id_concurrent_first_access_is_consistent() { let temp_dir = tempfile::tempdir().unwrap(); let path = temp_dir.path().join("db.sqlite"); // Create the datastore (and its schema) up front, then open two // separate connections to it, simulating two daemon RPC handlers // concurrently calling `Trident::new` (and therefore - // `datastore_id`) against the same datastore path. + // `installation_id`) against the same datastore path. super::DataStore::make_datastore(&path).unwrap(); let barrier = std::sync::Arc::new(std::sync::Barrier::new(2)); @@ -779,9 +966,118 @@ mod tests { std::thread::spawn(move || { let mut datastore = super::DataStore::open(&path).unwrap(); // Synchronize so both threads attempt "first access" - // (no database ID persisted yet) as close together + // (no installation ID persisted yet) as close together // as possible. barrier.wait(); + datastore.create_installation_id().unwrap() + }) + }) + .collect(); + + let ids: Vec = handles.into_iter().map(|h| h.join().unwrap()).collect(); + assert_eq!( + ids[0], ids[1], + "concurrent first access returned inconsistent installation IDs" + ); + + temp_dir.close().unwrap(); + } + + #[test] + fn test_installation_id_is_stable() { + let temp_dir = tempfile::tempdir().unwrap(); + let path = temp_dir.path().join("db.sqlite"); + let db = super::DataStore::make_datastore(&path).unwrap(); + let mut datastore = super::DataStore { + db: Some(db), + host_status: Default::default(), + temporary: false, + }; + + let id = datastore.create_installation_id().unwrap(); + // Calling installation_id again should return the same ID, not generate a + // new one. + assert_eq!(datastore.create_installation_id().unwrap(), id); + + temp_dir.close().unwrap(); + } + + #[test] + /// The read-only getter must not create an installation ID -- only + /// `create_installation_id` (called specifically at the start of + /// `Trident::install`) does that. + fn test_installation_id_read_only_getter_does_not_create() { + let temp_dir = tempfile::tempdir().unwrap(); + let path = temp_dir.path().join("db.sqlite"); + let db = super::DataStore::make_datastore(&path).unwrap(); + let mut datastore = super::DataStore { + db: Some(db), + host_status: Default::default(), + temporary: false, + }; + + assert_eq!( + datastore.installation_id().unwrap(), + None, + "no installation ID should exist before create_installation_id is called" + ); + + let created = datastore.create_installation_id().unwrap(); + + assert_eq!( + datastore.installation_id().unwrap(), + Some(created), + "the read-only getter should now see the created ID" + ); + + temp_dir.close().unwrap(); + } + + #[test] + /// Regression test: `new_servicing_id` always generates a *fresh* ID on + /// every call, unlike `create_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_new_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.new_servicing_id().unwrap(); + assert_eq!(datastore.servicing_id().unwrap(), Some(first)); + + let second = datastore.new_servicing_id().unwrap(); + assert_ne!( + first, second, + "new_servicing_id should generate a fresh ID on every call, not reuse the previous one" + ); + assert_eq!(datastore.servicing_id().unwrap(), Some(second)); + } + + #[test] + fn test_datastore_id_concurrent_first_access_is_consistent() { + let temp_dir = tempfile::tempdir().unwrap(); + let path = temp_dir.path().join("db.sqlite"); + + // Create the datastore (and its schema) up front, then open two + // separate connections to it, simulating two callers concurrently + // calling `datastore_id` against the same datastore path. + super::DataStore::make_datastore(&path).unwrap(); + + let barrier = std::sync::Arc::new(std::sync::Barrier::new(2)); + let handles: Vec<_> = (0..2) + .map(|_| { + let path = path.clone(); + let barrier = barrier.clone(); + std::thread::spawn(move || { + let mut datastore = super::DataStore::open(&path).unwrap(); + // Synchronize so both threads attempt "first access" + // (no database ID persisted yet) as close together as + // possible. + barrier.wait(); datastore.datastore_id().unwrap() }) }) @@ -890,6 +1186,26 @@ mod functional_test { ); } + #[functional_test] + fn test_installation_id_survives_persist() { + let temp_dir = TempDir::new().unwrap(); + let datastore_temp_path = temp_dir.path().join("db-tmp.sqlite"); + let datastore_path = temp_dir.path().join("db.sqlite"); + + // Generate a installation ID in the temporary datastore, then persist it. + let installation_id = { + let mut datastore = DataStore::open_or_create(&datastore_temp_path).unwrap(); + let installation_id = datastore.create_installation_id().unwrap(); + datastore.persist(&datastore_path).unwrap(); + installation_id + }; + + // Re-open the persisted datastore and verify the same installation ID is + // returned, rather than a new one being generated. + let mut datastore = DataStore::open(&datastore_path).unwrap(); + assert_eq!(datastore.create_installation_id().unwrap(), installation_id); + } + #[functional_test] fn test_datastore_id_survives_persist() { let temp_dir = TempDir::new().unwrap(); diff --git a/crates/trident/src/lib.rs b/crates/trident/src/lib.rs index bb3f62fb4..328c9b77a 100644 --- a/crates/trident/src/lib.rs +++ b/crates/trident/src/lib.rs @@ -56,8 +56,16 @@ pub use crate::{ }, grpc_client::client_main, logging::{ - background_log::BackgroundLog, background_uploader::BackgroundUploader, - logfwd::LogForwarder, logstream::Logstream, tracestream::TraceStream, + appinsights::AppInsightsSender, + background_log::BackgroundLog, + background_uploader::{BackgroundUploadHandle, BackgroundUploader}, + logfwd::LogForwarder, + logstream::Logstream, + operation_context::{ + run_with_captured_operation, run_with_operation, save_reboot_operation, + take_reboot_operation, OperationSource, + }, + tracestream::TraceStream, }, orchestrate::OrchestratorConnection, reboot::request_reboot_with_wait, @@ -82,6 +90,15 @@ lazy_static::lazy_static! { .expect("Failed to parse TRIDENT_VERSION as semver::Version"); } +/// Azure Monitor / Application Insights connection string, compiled in at +/// build time via the `AZURE_MONITOR_CONNECTION_STRING` environment +/// variable. Empty when the variable was not provided at build time. +pub const AZURE_MONITOR_CONNECTION_STRING: &str = + match option_env!("AZURE_MONITOR_CONNECTION_STRING") { + Some(v) => v, + None => "", + }; + /// Trident binary path. const TRIDENT_BINARY_PATH: &str = "/usr/bin/trident"; @@ -124,6 +141,10 @@ pub struct Trident { host_config: Option, orchestrator: Option, is_stream_image: bool, + /// Kept so `Trident::install` can attach a newly-created installation + /// ID to it -- see `DataStore::create_installation_id`, only ever + /// called from `install`/staging, unlike this constructor. + tracestream: TraceStream, } impl Trident { @@ -132,6 +153,43 @@ impl Trident { datastore_path: &Path, logstream: Logstream, tracestream: TraceStream, + ) -> Result { + Self::new_impl(config_source, datastore_path, logstream, tracestream, true) + } + + /// Identical to [`Self::new`], except it never attaches a persisted + /// `installation_id` to `tracestream` at construction time (this + /// invocation's earliest events, up to and including `trident_start`, + /// are simply left unattributed). + /// + /// Only meant for the CLI's multiboot install path (see `main.rs`): a + /// multiboot install on an already-provisioned host may go on to swap + /// to a brand-new temporary datastore inside `Trident::install` (see + /// there), distinct from the `datastore_path` given here (the + /// existing host's persistent datastore). Attaching *this* + /// constructor's installation ID -- the existing host's -- before + /// that swap decision is made would misattribute this invocation's + /// earliest telemetry to the wrong host installation for the + /// remainder of the run. `Trident::install` always attaches the + /// correct installation ID (of whichever datastore it ends up using) + /// before doing anything else, so it's safe, and strictly better, to + /// leave these few earliest events unattributed rather than guess + /// wrong. + pub fn new_deferring_installation_id( + config_source: Option, + datastore_path: &Path, + logstream: Logstream, + tracestream: TraceStream, + ) -> Result { + Self::new_impl(config_source, datastore_path, logstream, tracestream, false) + } + + fn new_impl( + config_source: Option, + datastore_path: &Path, + logstream: Logstream, + tracestream: TraceStream, + attach_installation_id: bool, ) -> Result { let host_config = config_source .map(|source| Self::load_host_config(&source)) @@ -207,23 +265,34 @@ impl Trident { info!("Running Trident in a container"); } - // Retrieve (or create, on first run) this host's unique - // database ID from the datastore actually used for servicing, - // and attach it to the shared TraceStream before any startup - // metrics are emitted, so every trace/metric -- including this - // very "trident_start" event -- carries it. This runs for every - // caller of `Trident::new` (both the CLI path and each daemon - // RPC handler), since they all supply `datastore_path`. - match DataStore::open_or_create(datastore_path).and_then(|mut ds| ds.datastore_id()) { - Ok(datastore_id) => { - info!("Datastore ID: {datastore_id}"); - tracestream.set_datastore_id(datastore_id.to_string()); - } - Err(e) => { - warn!("Failed to get or create database ID: {e:?}"); - } + // Attach this host's installation ID -- if one 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 it once available. Read-only: + // the installation ID is only ever *created* by `Trident::install` + // (at the start of staging), never here, 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. + // + // Skipped entirely 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. + if attach_installation_id { + tracestream.attach_installation_id_if_present(datastore_path); } + // Attach this datastore's database ID (stable for its entire + // lifetime, unlike installation_id) whenever a datastore already + // exists at `datastore_path` -- never creates one. Not gated on + // `attach_installation_id`/multiboot: a multiboot install's + // temporary datastore does not exist yet at this point, so this + // is a no-op until the datastore is actually created/opened + // further down (see the `datastore_id()` calls near + // `create_and_attach_installation_id` below). + tracestream.attach_datastore_id_if_present(datastore_path); + // Trace features enabled in the Host Configuration. if let Some(hc) = &host_config { hc.feature_tracing(); @@ -250,6 +319,7 @@ impl Trident { host_config, orchestrator, is_stream_image: false, + tracestream, }) } @@ -459,6 +529,7 @@ impl Trident { ))?; let is_stream_image = self.is_stream_image; + let tracestream = self.tracestream.clone(); self.execute_and_record_error(datastore, |datastore| { host_config @@ -503,6 +574,34 @@ impl Trident { } } + // Create (or read back, if one already exists) this install's + // installation ID and attach it to the shared TraceStream, now + // that the multiboot swap above (if any) has settled on the + // datastore this install actually uses -- see the doc comment + // on `new_deferring_installation_id`. Uses `datastore` as it + // stands after any multiboot swap above, so a multiboot + // install's own (new, eventually-persistent) datastore gets + // its own installation ID, not the already-provisioned host's. + // Best-effort: a failure here must not block the install + // itself, since telemetry attribution is not load-bearing for + // servicing outcomes (same invariant `update`'s CIH bootstrap + // path already honors below). + if let Err(e) = tracestream.create_and_attach_installation_id(datastore) { + warn!("Failed to create installation ID: {e:?}"); + } + + // Get (or, for a brand-new datastore, create) this datastore's + // database ID and attach it. Best-effort, same rationale as + // installation ID above: telemetry attribution must never + // block servicing. + match datastore.datastore_id() { + Ok(datastore_id) => { + info!("Datastore ID: {datastore_id}"); + tracestream.set_datastore_id(datastore_id.to_string()); + } + Err(e) => warn!("Failed to get/create database ID: {e:?}"), + } + // Use a prefetched image if provided, otherwise load the image // specified in the Host Configuration. let image = match prefetched_image { @@ -599,6 +698,8 @@ impl Trident { "update called without Host Configuration set", ))?; + let tracestream = self.tracestream.clone(); + self.execute_and_record_error(datastore, |datastore| { // Ensure that the datastore exists. if !datastore.is_persistent() { @@ -614,6 +715,30 @@ impl Trident { status.is_management_os = false; }) .message("Failed to initialize datastore")?; + + // This host has just adopted a datastore via the CIH + // bootstrap path, bypassing `Trident::install` entirely + // -- nothing else will ever create an installation ID for + // it otherwise. Best-effort: a failure here must not + // block the update itself, since telemetry attribution + // is not load-bearing for servicing outcomes. Known + // remaining gap: this invocation's own + // command_start/trident_start (fired even earlier, in + // the CLI/daemon dispatch and Trident::new respectively) + // still won't carry it -- both fire before this point. + if let Err(e) = tracestream.create_and_attach_installation_id(datastore) { + warn!("Failed to create installation ID during CIH bootstrap: {e:?}"); + } + + match datastore.datastore_id() { + Ok(datastore_id) => { + info!("Datastore ID: {datastore_id}"); + tracestream.set_datastore_id(datastore_id.to_string()); + } + Err(e) => { + warn!("Failed to get/create database ID during CIH bootstrap: {e:?}") + } + } } else { // For non-CIH images, if the datastore is not persistent, return error return Err(TridentError::new(InvalidInputError::HostNotProvisioned)) diff --git a/crates/trident/src/logging/appinsights.rs b/crates/trident/src/logging/appinsights.rs new file mode 100644 index 000000000..3a61fb29c --- /dev/null +++ b/crates/trident/src/logging/appinsights.rs @@ -0,0 +1,780 @@ +//! Best-effort tracing sink that forwards Trident's metric/span tracing +//! events to Azure Monitor / Application Insights. +//! +//! This intentionally does not depend on the OpenTelemetry SDK or an +//! Application Insights client crate. It follows the same minimal approach +//! as [`super::tracestream::TraceSender`]: parse the Application Insights +//! *connection string* (`InstrumentationKey=;IngestionEndpoint=;...`) +//! ourselves and build the raw Application Insights `EventData` envelope. +//! +//! Sending is delegated to the same [`super::background_uploader`] used by +//! [`super::logstream::Logstream`]: `send_event` only enqueues the envelope +//! and returns immediately, so tracing-layer callbacks (which run on +//! whichever thread emitted the event) are never blocked on network I/O. +//! The background uploader performs the actual `POST` to +//! `${ingestion_endpoint}/v2/track` with a short, bounded timeout on its own +//! dedicated thread. Failures (enqueue, network, non-2xx, etc.) are +//! logged and otherwise swallowed -- telemetry must never be able to affect +//! servicing outcomes. + +use std::{ + collections::BTreeMap, + sync::{Arc, RwLock}, + time::{Duration, Instant}, +}; + +use anyhow::Context; +use log::trace; +use serde_json::{json, Value}; +use tracing::{ + field::{Field, Visit}, + span, Event, Subscriber, +}; +use tracing_subscriber::{layer::Layer, registry::LookupSpan}; +use url::Url; + +use super::{ + background_uploader::BackgroundUploadHandle, + tracestream::{merge_operation_context, PLATFORM_INFO}, +}; +use crate::TRIDENT_VERSION; + +/// Default Application Insights ingestion endpoint, used when the connection +/// string does not specify one explicitly. +const DEFAULT_INGESTION_ENDPOINT: &str = "https://dc.services.visualstudio.com"; + +/// Per-request total timeout, enforced by the background uploader. Telemetry +/// must never meaningfully delay Trident's actual work. +const REQUEST_TIMEOUT: Duration = Duration::from_secs(5); + +/// `Content-Type` for the Application Insights ingestion request. +const CONTENT_TYPE_JSON: &str = "application/json"; + +/// A parsed Application Insights connection string. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ConnParts { + /// Ingestion endpoint, trailing slash stripped (e.g. + /// `https://region.in.applicationinsights.azure.com`). + pub ingestion_endpoint: String, + /// Instrumentation key. + pub instrumentation_key: String, +} + +impl ConnParts { + /// The `POST` target: `${ingestion_endpoint}/v2/track`. + fn track_url(&self) -> Option { + Url::parse(&format!( + "{}/v2/track", + self.ingestion_endpoint.trim_end_matches('/') + )) + .ok() + } +} + +/// Parse an Application Insights connection string of the form +/// `InstrumentationKey=;IngestionEndpoint=https://...;...`. Returns `None` +/// if the string is empty, unparsable, or missing an instrumentation key. +/// +/// If `IngestionEndpoint` is not given explicitly, the endpoint is derived +/// from the sovereign-cloud `EndpointSuffix`/`Location` fields when present +/// (e.g. `EndpointSuffix=applicationinsights.azure.cn;Location=chinaeast2` -> +/// `https://chinaeast2.dc.applicationinsights.azure.cn`), matching the +/// Azure Monitor SDKs' documented connection-string format, which always +/// uses the `dc` ingestion prefix (with or without `Location`) -- not `in`. +/// Only if none of `IngestionEndpoint`/`EndpointSuffix` are present does +/// this fall back to the public Application Insights endpoint -- a +/// sovereign-cloud string must never be silently redirected to the public +/// endpoint. +pub(crate) fn parse_connection_string(s: &str) -> Option { + let mut instrumentation_key: Option = None; + let mut ingestion_endpoint: Option = None; + let mut endpoint_suffix: Option = None; + let mut location: Option = None; + + for part in s.split(';') { + let part = part.trim(); + if part.is_empty() { + continue; + } + let Some((key, value)) = part.split_once('=') else { + continue; + }; + match key.trim().to_ascii_lowercase().as_str() { + "instrumentationkey" => instrumentation_key = Some(value.trim().to_string()), + "ingestionendpoint" => { + ingestion_endpoint = Some(value.trim().trim_end_matches('/').to_string()) + } + "endpointsuffix" => endpoint_suffix = Some(value.trim().trim_matches('/').to_string()), + "location" => location = Some(value.trim().to_string()), + _ => {} + } + } + + let instrumentation_key = instrumentation_key.filter(|k| !k.is_empty())?; + + let ingestion_endpoint = match ingestion_endpoint.filter(|e| !e.is_empty()) { + Some(explicit) => explicit, + None => match endpoint_suffix.filter(|s| !s.is_empty()) { + // Sovereign-cloud form: derive the ingestion endpoint from + // EndpointSuffix (+ optional Location), rather than assuming the + // public endpoint. The ingestion prefix is always `dc` + // (matching the public endpoint's own + // `dc.services.visualstudio.com` shape), with or without + // Location -- not `in`, which the Azure Monitor SDKs never + // produce and which would target a nonstandard/nonexistent + // host for sovereign clouds. + Some(suffix) => match location.filter(|l| !l.is_empty()) { + Some(location) => format!("https://{location}.dc.{suffix}"), + None => format!("https://dc.{suffix}"), + }, + None => DEFAULT_INGESTION_ENDPOINT.to_string(), + }, + }; + + Some(ConnParts { + ingestion_endpoint, + instrumentation_key, + }) +} + +/// A visitor that records the fields of a tracing event/span as a +/// `BTreeMap`, mirroring [`super::tracestream::TraceEntryVisitor`]. +#[derive(Default)] +struct FieldVisitor { + fields: BTreeMap, +} + +impl Visit for FieldVisitor { + fn record_i64(&mut self, field: &Field, value: i64) { + self.fields.insert(field.name().to_string(), json!(value)); + } + + fn record_f64(&mut self, field: &Field, value: f64) { + self.fields.insert(field.name().to_string(), json!(value)); + } + + fn record_u64(&mut self, field: &Field, value: u64) { + self.fields.insert(field.name().to_string(), json!(value)); + } + + fn record_bool(&mut self, field: &Field, value: bool) { + self.fields.insert(field.name().to_string(), json!(value)); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.fields.insert(field.name().to_string(), json!(value)); + } + + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.fields + .insert(field.name().to_string(), json!(format!("{value:?}"))); + } +} + +/// Timestamp recorded when a span is entered, used to compute execution time +/// on exit. +struct SpanStart(Instant); + +/// Renders a JSON value as a string, since Application Insights `EventData` +/// properties are a `Map`. +fn stringify(value: &Value) -> String { + match value { + Value::String(s) => s.clone(), + other => other.to_string(), + } +} + +/// A `tracing_subscriber::Layer` that forwards Trident's metric events and +/// instrumented spans to Application Insights, best-effort. +/// +/// Only constructed when a connection string was compiled into the binary +/// (see [`crate::AZURE_MONITOR_CONNECTION_STRING`]) *and* telemetry has been +/// enabled via the Agent Configuration file (see +/// [`crate::agentconfig::AgentConfig::telemetry_enabled`]); see +/// [`AppInsightsSender::from_connection_string`]. +pub struct AppInsightsSender { + instrumentation_key: String, + track_url: Url, + uploader: BackgroundUploadHandle, + /// The same persistent, per-host installation ID handle used by + /// `TraceStream`/`TraceSender` (see `TraceStream::installation_id_handle`), + /// so Application Insights events can be correlated back to a specific + /// host installation the same way tracestream metrics already are. + installation_id: Arc>>, + /// The same persistent database ID handle used by `TraceStream`/ + /// `TraceSender` (see `TraceStream::datastore_id_handle`), stable for + /// the datastore's entire lifetime rather than a single install. + datastore_id: Arc>>, +} + +impl AppInsightsSender { + /// Build a sender from an Application Insights connection string. + /// Returns `None` if the string is empty, fails to parse, the + /// ingestion endpoint does not form a valid URL, or that URL's scheme + /// is not `https`. + /// + /// The events sent through this sender include host identifiers (see + /// [`super::tracestream::PLATFORM_INFO`]), so a non-HTTPS endpoint -- + /// e.g. from a build-time typo/misconfiguration -- is rejected rather + /// than silently sending opted-in host telemetry in cleartext. Azure + /// Monitor ingestion endpoints require HTTPS. + pub fn from_connection_string( + connection_string: &str, + uploader: BackgroundUploadHandle, + installation_id: Arc>>, + datastore_id: Arc>>, + ) -> Option { + let parts = parse_connection_string(connection_string)?; + match parts.track_url() { + Some(url) if url.scheme() == "https" => {} + _ => { + trace!( + "Application Insights ingestion endpoint '{}' is not HTTPS, disabling telemetry", + parts.ingestion_endpoint + ); + return None; + } + } + Self::from_parts(parts, uploader, installation_id, datastore_id) + } + + fn from_parts( + parts: ConnParts, + uploader: BackgroundUploadHandle, + installation_id: Arc>>, + datastore_id: Arc>>, + ) -> Option { + let track_url = parts.track_url()?; + Some(Self { + instrumentation_key: parts.instrumentation_key, + track_url, + uploader, + installation_id, + datastore_id, + }) + } + + /// Build and enqueue an Application Insights `EventData` envelope for + /// the background uploader to send, best-effort. This only serializes + /// the envelope and hands it off to the uploader's channel, so it never + /// blocks on network I/O. A serialization failure or a closed uploader + /// is logged here at `trace` level; a later network error or non-2xx + /// response is logged by the background uploader itself (at `error` + /// level, same as any other background upload). Either way the failure + /// is otherwise ignored -- it can never affect servicing outcomes. + fn send_event(&self, name: &str, mut properties: BTreeMap) { + properties.insert("trident_version".to_string(), json!(TRIDENT_VERSION)); + for (key, value) in PLATFORM_INFO.iter() { + properties.insert(key.clone(), json!(stringify(value))); + } + if let Ok(installation_id) = self.installation_id.read() { + if let Some(installation_id) = installation_id.as_ref() { + properties + .entry("installation_id".to_string()) + .or_insert_with(|| json!(installation_id)); + } + } + if let Ok(datastore_id) = self.datastore_id.read() { + if let Some(datastore_id) = datastore_id.as_ref() { + properties + .entry("datastore_id".to_string()) + .or_insert_with(|| json!(datastore_id)); + } + } + // operation_id/command/installation_id-fallback enrichment is + // shared with the local metrics-file sink -- see + // `merge_operation_context`'s doc comment for why. + merge_operation_context(&mut properties); + + let string_properties: BTreeMap = properties + .into_iter() + .map(|(key, value)| (key, stringify(&value))) + .collect(); + + let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true); + let envelope = json!({ + "name": format!( + "Microsoft.ApplicationInsights.{}.Event", + self.instrumentation_key + ), + "time": now, + "iKey": self.instrumentation_key, + "tags": { "ai.internal.sdkVersion": format!("trident:{TRIDENT_VERSION}") }, + "data": { + "baseType": "EventData", + "baseData": { + "ver": 2, + "name": name, + "properties": string_properties, + } + } + }); + + let body = match serde_json::to_vec(&envelope) { + Ok(b) => b, + Err(e) => { + trace!("Failed to serialize Application Insights event: {e}"); + return; + } + }; + + if let Err(e) = self.uploader.upload_with_validator( + &self.track_url, + body, + REQUEST_TIMEOUT, + Some(CONTENT_TYPE_JSON), + Some(validate_track_response), + ) { + trace!("Failed to enqueue Application Insights event: {e}"); + } + } +} + +/// Response validator for the Application Insights `/v2/track` endpoint +/// (see [`BackgroundUploadHandle::upload_with_validator`]). A 2xx status +/// alone is not sufficient here: the endpoint returns 206 Partial Success +/// when only some of the submitted items were accepted, with an +/// `itemsReceived`/`itemsAccepted` body. Since every request from this +/// sender carries exactly one envelope, the only response that actually +/// means "accepted" is `itemsReceived == 1 && itemsAccepted == 1` -- +/// checking mere equality between the two counts would also accept +/// impossible pairs for a one-envelope request (e.g. `0/0` or `2/2`), +/// silently clearing backoff even though the event wasn't received as +/// expected. Anything other than exactly `1/1` is treated as a failure so +/// it goes through the same retry/backoff path as a network-level error, +/// instead of being silently discarded as a false "success". +fn validate_track_response(status: reqwest::StatusCode, body: &[u8]) -> Result<(), anyhow::Error> { + if status != reqwest::StatusCode::PARTIAL_CONTENT { + return Ok(()); + } + + let parsed: Value = serde_json::from_slice(body) + .context("Failed to parse Application Insights partial-success response body")?; + let items_received = parsed.get("itemsReceived").and_then(Value::as_u64); + let items_accepted = parsed.get("itemsAccepted").and_then(Value::as_u64); + + match (items_received, items_accepted) { + (Some(1), Some(1)) => Ok(()), + (Some(received), Some(accepted)) => anyhow::bail!( + "Application Insights accepted only {accepted} of {received} submitted items \ + (expected exactly 1 of 1 for this single-envelope request)" + ), + _ => anyhow::bail!( + "Application Insights returned 206 Partial Success without a parsable \ + itemsReceived/itemsAccepted body: {}", + String::from_utf8_lossy(body) + ), + } +} + +/// The `Layer` implementation mirrors +/// [`super::tracestream::TraceSender`]'s event/span handling, but renders an +/// Application Insights `EventData` envelope instead of Trident's own +/// metrics-file format. +impl Layer for AppInsightsSender +where + S: Subscriber + for<'span> LookupSpan<'span>, +{ + fn enabled( + &self, + metadata: &tracing::Metadata<'_>, + _ctx: tracing_subscriber::layer::Context<'_, S>, + ) -> bool { + metadata.level() <= &tracing::Level::INFO + } + + fn on_event(&self, event: &Event<'_>, _ctx: tracing_subscriber::layer::Context<'_, S>) { + let mut visitor = FieldVisitor::default(); + event.record(&mut visitor); + + let Some(metric_name) = visitor + .fields + .get("metric_name") + .and_then(|v| v.as_str()) + .map(str::to_string) + else { + // Not a metric event (e.g. a plain log line); nothing to forward. + return; + }; + + let properties: BTreeMap = visitor + .fields + .into_iter() + .filter(|(key, _)| key != "metric_name") + .collect(); + + self.send_event(&metric_name, properties); + } + + fn on_new_span( + &self, + attrs: &span::Attributes<'_>, + id: &span::Id, + ctx: tracing_subscriber::layer::Context<'_, S>, + ) { + if let Some(span) = ctx.span(id) { + let mut visitor = FieldVisitor::default(); + attrs.record(&mut visitor); + span.extensions_mut().insert(visitor); + } + } + + fn on_enter(&self, id: &span::Id, ctx: tracing_subscriber::layer::Context<'_, S>) { + if let Some(span) = ctx.span(id) { + span.extensions_mut().insert(SpanStart(Instant::now())); + } + } + + fn on_exit(&self, id: &span::Id, ctx: tracing_subscriber::layer::Context<'_, S>) { + let Some(span) = ctx.span(id) else { + return; + }; + let Some(SpanStart(start)) = span.extensions_mut().remove::() else { + return; + }; + let Some(mut visitor) = span.extensions_mut().remove::() else { + return; + }; + + visitor.fields.insert( + "execution_time".to_string(), + json!(start.elapsed().as_secs_f64()), + ); + + self.send_event(span.name(), visitor.fields); + } + + fn on_record( + &self, + id: &span::Id, + values: &span::Record<'_>, + ctx: tracing_subscriber::layer::Context<'_, S>, + ) { + if let Some(span) = ctx.span(id) { + if let Some(visitor) = span.extensions_mut().get_mut::() { + values.record(visitor); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_validate_track_response_non_206_ignores_body() { + // Any non-206 2xx status is only ever reached via the caller's + // own `is_success()` check, so the validator doesn't need to + // (and shouldn't) inspect the body for it. + assert!(validate_track_response(reqwest::StatusCode::OK, b"not json at all").is_ok()); + } + + #[test] + fn test_validate_track_response_206_all_items_accepted() { + let body = br#"{"itemsReceived":1,"itemsAccepted":1}"#; + assert!(validate_track_response(reqwest::StatusCode::PARTIAL_CONTENT, body).is_ok()); + } + + #[test] + fn test_validate_track_response_206_item_rejected() { + let body = br#"{"itemsReceived":1,"itemsAccepted":0}"#; + let err = validate_track_response(reqwest::StatusCode::PARTIAL_CONTENT, body) + .expect_err("a rejected single-envelope request should be treated as a failure"); + assert!(err.to_string().contains("accepted only 0 of 1")); + } + + #[test] + fn test_validate_track_response_206_zero_zero_is_not_success() { + // A malformed/impossible response for a single-envelope request -- + // mere equality between the two counts (0 == 0) must NOT be + // mistaken for success. + let body = br#"{"itemsReceived":0,"itemsAccepted":0}"#; + let err = validate_track_response(reqwest::StatusCode::PARTIAL_CONTENT, body) + .expect_err("itemsReceived=0/itemsAccepted=0 must not be treated as success"); + assert!(err.to_string().contains("accepted only 0 of 0")); + } + + #[test] + fn test_validate_track_response_206_impossible_counts_above_one() { + // Also impossible for a single-envelope request -- equality + // alone (2 == 2) must not be mistaken for success either. + let body = br#"{"itemsReceived":2,"itemsAccepted":2}"#; + let err = validate_track_response(reqwest::StatusCode::PARTIAL_CONTENT, body) + .expect_err("itemsReceived=2/itemsAccepted=2 is impossible for a 1-item request"); + assert!(err.to_string().contains("accepted only 2 of 2")); + } + + #[test] + fn test_validate_track_response_206_unparsable_body() { + let err = validate_track_response(reqwest::StatusCode::PARTIAL_CONTENT, b"not json") + .expect_err( + "an unparsable 206 body should be treated as a failure, not silently accepted", + ); + assert!(err.to_string().contains("Failed to parse")); + } + + #[test] + fn test_parse_connection_string_full() { + let parts = parse_connection_string( + "InstrumentationKey=abc123;IngestionEndpoint=https://region.example/;LiveEndpoint=https://live.example/", + ) + .expect("should parse"); + assert_eq!(parts.instrumentation_key, "abc123"); + assert_eq!(parts.ingestion_endpoint, "https://region.example"); + assert_eq!( + parts.track_url(), + Some(Url::parse("https://region.example/v2/track").unwrap()) + ); + } + + #[test] + fn test_parse_connection_string_missing_endpoint_uses_default() { + let parts = parse_connection_string("InstrumentationKey=abc123").expect("should parse"); + assert_eq!(parts.instrumentation_key, "abc123"); + assert_eq!(parts.ingestion_endpoint, DEFAULT_INGESTION_ENDPOINT); + assert_eq!( + parts.track_url(), + Some(Url::parse(&format!("{DEFAULT_INGESTION_ENDPOINT}/v2/track")).unwrap()) + ); + } + + #[test] + /// Sovereign-cloud connection strings that specify `EndpointSuffix` (and + /// optionally `Location`) instead of `IngestionEndpoint` must derive the + /// matching sovereign ingestion endpoint, not silently fall back to the + /// public one. + fn test_parse_connection_string_endpoint_suffix_with_location() { + let parts = parse_connection_string( + "InstrumentationKey=abc123;EndpointSuffix=applicationinsights.azure.cn;Location=chinaeast2", + ) + .expect("should parse"); + assert_eq!(parts.instrumentation_key, "abc123"); + assert_eq!( + parts.ingestion_endpoint, + "https://chinaeast2.dc.applicationinsights.azure.cn" + ); + } + + #[test] + fn test_parse_connection_string_endpoint_suffix_without_location() { + let parts = parse_connection_string( + "InstrumentationKey=abc123;EndpointSuffix=applicationinsights.azure.cn", + ) + .expect("should parse"); + assert_eq!( + parts.ingestion_endpoint, + "https://dc.applicationinsights.azure.cn" + ); + } + + #[test] + fn test_parse_connection_string_explicit_ingestion_endpoint_wins_over_suffix() { + let parts = parse_connection_string( + "InstrumentationKey=abc123;IngestionEndpoint=https://region.example/;EndpointSuffix=applicationinsights.azure.cn", + ) + .expect("should parse"); + assert_eq!(parts.ingestion_endpoint, "https://region.example"); + } + + #[test] + fn test_parse_connection_string_missing_ikey_is_none() { + assert!(parse_connection_string("IngestionEndpoint=https://region.example/").is_none()); + assert!(parse_connection_string( + "InstrumentationKey=;IngestionEndpoint=https://region.example/" + ) + .is_none()); + } + + #[test] + fn test_parse_connection_string_empty_is_none() { + assert!(parse_connection_string("").is_none()); + } + + #[test] + fn test_from_connection_string_empty_is_none() { + assert!(AppInsightsSender::from_connection_string( + "", + BackgroundUploadHandle::new_mock(), + Arc::new(RwLock::new(None)), + Arc::new(RwLock::new(None)), + ) + .is_none()); + } + + #[test] + fn test_from_connection_string_builds_sender() { + let sender = AppInsightsSender::from_connection_string( + "InstrumentationKey=k;IngestionEndpoint=https://region.example/", + BackgroundUploadHandle::new_mock(), + Arc::new(RwLock::new(None)), + Arc::new(RwLock::new(None)), + ) + .expect("should build sender"); + assert_eq!(sender.instrumentation_key, "k"); + assert_eq!( + sender.track_url, + Url::parse("https://region.example/v2/track").unwrap() + ); + } + + #[test] + /// A non-HTTPS ingestion endpoint (e.g. from a build-time typo/ + /// misconfiguration) must be rejected rather than silently accepted -- + /// events sent through this sender include host identifiers. + fn test_from_connection_string_rejects_non_https_endpoint() { + assert!(AppInsightsSender::from_connection_string( + "InstrumentationKey=k;IngestionEndpoint=http://region.example/", + BackgroundUploadHandle::new_mock(), + Arc::new(RwLock::new(None)), + Arc::new(RwLock::new(None)), + ) + .is_none()); + } + + #[test] + fn test_stringify() { + assert_eq!(stringify(&json!("hello")), "hello"); + assert_eq!(stringify(&json!(42)), "42"); + assert_eq!(stringify(&json!(true)), "true"); + } +} + +#[cfg(feature = "functional-test")] +#[cfg_attr(not(test), allow(unused_imports, dead_code))] +mod functional_test { + use super::*; + // Only used by this feature-gated module (a plain `cargo check`/`cargo + // test` without `--features functional-test` never compiles this mod, + // which would otherwise make the top-level import unused). + use crate::logging::operation_context; + + use std::{ + io::{Read, Write}, + net::{TcpListener, TcpStream}, + sync::mpsc::channel, + }; + + use pytest_gen::functional_test; + use tracing_subscriber::{filter, layer::SubscriberExt}; + + /// Reads a full HTTP request off `stream`. A single `TcpStream::read` + /// call is not guaranteed to return the entire request (TCP is a byte + /// stream, not message-oriented) -- keep reading until the header + /// terminator has arrived and, per the declared `Content-Length`, the + /// full body has too. + fn read_full_http_request(stream: &mut TcpStream) -> String { + let mut buf: Vec = Vec::new(); + let mut chunk = [0u8; 4096]; + + loop { + match stream.read(&mut chunk) { + Ok(0) => break, // Peer closed the connection. + Ok(n) => buf.extend_from_slice(&chunk[..n]), + Err(_) => break, + } + + let Some(headers_end) = find_subslice(&buf, b"\r\n\r\n") else { + continue; // Headers not fully received yet. + }; + let headers = String::from_utf8_lossy(&buf[..headers_end]); + let content_length: usize = headers + .lines() + .find_map(|line| { + line.to_lowercase() + .strip_prefix("content-length:") + .map(|v| v.trim().to_string()) + }) + .and_then(|v| v.parse().ok()) + .unwrap_or(0); + + let body_end = headers_end + content_length; + if buf.len() >= body_end { + buf.truncate(body_end); + break; + } + } + + String::from_utf8_lossy(&buf).to_string() + } + + fn find_subslice(haystack: &[u8], needle: &[u8]) -> Option { + haystack + .windows(needle.len()) + .position(|window| window == needle) + .map(|pos| pos + needle.len()) + } + + /// Spins up a local TCP listener standing in for the Application + /// Insights ingestion endpoint, and confirms the sender actually posts a + /// well-formed `EventData` envelope to it over the network. + #[functional_test] + fn test_app_insights_sender_posts_event() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + + // command_start (fired by run_with_operation) and test_metric below + // are each posted as their own request, so accept and collect every + // connection the listener sees rather than assuming exactly one. + let (tx, rx) = channel(); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { break }; + let received = read_full_http_request(&mut stream); + // Send a minimal response so the client's request completes + // cleanly instead of hitting a connection reset. + let _ = stream.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"); + if tx.send(received).is_err() { + break; + } + } + }); + + let uploader = crate::BackgroundUploader::new().expect("should build uploader"); + let sender = AppInsightsSender::from_parts( + ConnParts { + ingestion_endpoint: format!("http://{addr}"), + instrumentation_key: "test-key".to_string(), + }, + uploader.get_handle().expect("uploader should be alive"), + Arc::new(RwLock::new(Some("test-installation-id".to_string()))), + Arc::new(RwLock::new(Some("test-datastore-id".to_string()))), + ) + .expect("should build sender") + .with_filter(filter::LevelFilter::INFO); + + let _guard = + tracing::subscriber::set_default(tracing_subscriber::Registry::default().with(sender)); + + // Wrapping in run_with_operation confirms operation_id/command/source + // also reach the outgoing properties, alongside installation_id above. + operation_context::run_with_operation( + "test_command", + operation_context::OperationSource::Cli, + || { + tracing::info!(metric_name = "test_metric", value = true); + }, + ); + + // Collect both requests (command_start + test_metric); order between + // them is not guaranteed, so gather everything seen within the + // timeout and assert across the combined traffic. + let mut requests = Vec::new(); + while requests.len() < 2 { + match rx.recv_timeout(std::time::Duration::from_secs(5)) { + Ok(req) => requests.push(req), + Err(e) => panic!("did not receive the expected requests: {e:?}"), + } + } + let combined = requests.join("\n"); + + assert!(combined.contains("POST /v2/track")); + assert!(combined.contains("\"name\":\"test_metric\"")); + assert!(combined.contains("\"iKey\":\"test-key\"")); + assert!(combined.contains("\"installation_id\":\"test-installation-id\"")); + assert!(combined.contains("\"datastore_id\":\"test-datastore-id\"")); + assert!(combined.contains("\"command\":\"test_command\"")); + assert!(combined.contains("\"operation_id\":")); + assert!(combined.contains("\"source\":\"cli\"")); + } +} diff --git a/crates/trident/src/logging/background_uploader.rs b/crates/trident/src/logging/background_uploader.rs index 9dd3914d7..962695f5c 100644 --- a/crates/trident/src/logging/background_uploader.rs +++ b/crates/trident/src/logging/background_uploader.rs @@ -1,8 +1,8 @@ use std::{ - collections::HashSet, + collections::HashMap, sync::LazyLock, thread::{Builder, JoinHandle}, - time::Duration, + time::{Duration, Instant}, }; use anyhow::{bail, Context, Error}; @@ -20,11 +20,63 @@ static HTTP_ASYNC_CLIENT: LazyLock = LazyLock::new(Client::new); /// The module path of the background uploader. Can be used for filtering logs. pub(super) const BACKGROUND_LOG_MODULE: &str = module_path!(); +/// Cooldown applied after an origin's first consecutive failure, doubled +/// for each further consecutive failure (see [`OriginCooldown`]) up to +/// [`MAX_ORIGIN_COOLDOWN`]. Bounding the backoff instead of disabling the +/// origin outright means a healthy endpoint recovers on its own after a +/// transient blip (a momentary network hiccup, a brief server restart), +/// while a genuinely dead one is still backed off hard enough not to waste +/// effort retrying it constantly. +const BASE_ORIGIN_COOLDOWN: Duration = Duration::from_secs(30); + +/// Upper bound on the exponential backoff described above. +const MAX_ORIGIN_COOLDOWN: Duration = Duration::from_secs(600); + +/// Validates an upload response beyond a bare 2xx status, for callers whose +/// ingestion protocol can reject part of a request while still returning a +/// 2xx status (e.g. Application Insights' 206 Partial Success). Given the +/// response status and body, returns `Ok(())` if the upload should be +/// treated as a success, or `Err` (triggering the same retry/backoff path +/// as a network-level failure) otherwise. +pub(crate) type ResponseValidator = fn(reqwest::StatusCode, &[u8]) -> Result<(), Error>; + /// Data to be uploaded by the background uploader. struct UploadData { url: Url, body: Vec, timeout: Duration, + /// Optional `Content-Type` header value to attach to the request. + content_type: Option<&'static str>, + /// Optional response validator -- see [`ResponseValidator`]. When + /// `None`, falls back to treating any 2xx status as success. + response_validator: Option, +} + +/// Per-origin backoff state, tracked across consecutive failed uploads to +/// the same origin. Reset (removed from the tracking map) as soon as an +/// upload to that origin succeeds again. +struct OriginCooldown { + /// Consecutive failures observed for this origin since its last + /// success (or since tracking began). + consecutive_failures: u32, + /// Uploads to this origin are skipped until this instant. + cooldown_until: Instant, +} + +/// Computes the exponential backoff duration for the given number of +/// consecutive failures (1 for the first failure, 2 for the second, ...), +/// doubling from `BASE_ORIGIN_COOLDOWN` and clamped at +/// `MAX_ORIGIN_COOLDOWN`. Split out from `upload_loop` so the pure +/// calculation is independently testable without needing to simulate real +/// time passing. +fn backoff_for_failures(consecutive_failures: u32) -> Duration { + // Cap the exponent well below any value that could overflow the + // shift: MAX_ORIGIN_COOLDOWN already clamps the result, so this only + // needs to be large enough to reach that clamp. + let exponent = consecutive_failures.saturating_sub(1).min(16); + BASE_ORIGIN_COOLDOWN + .saturating_mul(1u32 << exponent) + .min(MAX_ORIGIN_COOLDOWN) } /// A background uploader that sends log data to a remote server asynchronously. @@ -86,41 +138,153 @@ impl BackgroundUploader { /// The main upload loop that processes incoming upload requests. async fn upload_loop(mut receiver: UnboundedReceiver) { - let mut ignored_servers = HashSet::new(); + let mut origin_cooldowns: HashMap = HashMap::new(); while let Some(upload) = receiver.recv().await { - if ignored_servers.contains(&upload.url.origin()) { - continue; + let origin = upload.url.origin(); + + if let Some(state) = origin_cooldowns.get(&origin) { + if Instant::now() < state.cooldown_until { + continue; + } } - let result = HTTP_ASYNC_CLIENT + let mut request = HTTP_ASYNC_CLIENT .post(upload.url.clone()) .timeout(upload.timeout) - .body(upload.body) - .send() - .await; - - if let Err(e) = result { - error!("Background upload failed: {e}"); - ignored_servers.insert(upload.url.origin()); - error!( - "Ignoring future uploads to server: {}", - match upload.url.origin() { - Origin::Tuple(scheme, host, port) => - format!("{}://{}:{}", scheme, host, port), - Origin::Opaque(_) => "[opaque origin]".to_string(), + .body(upload.body); + if let Some(content_type) = upload.content_type { + request = request.header(reqwest::header::CONTENT_TYPE, content_type); + } + // Treat non-2xx responses the same as a network-level failure: a + // consumer (e.g. AppInsightsSender) may document that rejected + // requests count as failures, so surface them here rather than + // silently treating any response as success. `error_for_status()` + // alone is not enough: it only rejects 4xx/5xx, so a 3xx (e.g. an + // unexpected redirect the client never followed) would still be + // reported as success. Explicitly require 2xx instead. A 2xx + // status alone is still not sufficient for every caller: some + // ingestion protocols (e.g. Application Insights) can return a + // 2xx (206 Partial Success) while rejecting part or all of the + // request body, so a caller-supplied `response_validator` gets + // the final say when present. + let response_validator = upload.response_validator; + let result: Result<(), Error> = match request.send().await { + Ok(response) if response.status().is_success() => { + let status = response.status(); + match response_validator { + Some(validate) => match response.bytes().await { + Ok(body) => validate(status, &body), + Err(e) => Err(e.into()), + }, + None => Ok(()), } - ); + } + Ok(response) => Err(anyhow::anyhow!( + "unexpected HTTP status {} from {}", + response.status(), + response.url() + )), + Err(e) => Err(e.into()), + }; + + match result { + Ok(()) => { + // Origin is healthy again: drop any backoff state so a + // future failure starts from the base cooldown rather + // than a previously-escalated one. + origin_cooldowns.remove(&origin); + } + Err(e) => { + error!("Background upload failed: {e}"); + + let consecutive_failures = origin_cooldowns + .get(&origin) + .map(|state| state.consecutive_failures) + .unwrap_or(0) + + 1; + let cooldown = backoff_for_failures(consecutive_failures); + let cooldown_until = Instant::now() + cooldown; + + origin_cooldowns.insert( + origin.clone(), + OriginCooldown { + consecutive_failures, + cooldown_until, + }, + ); + + error!( + "Backing off uploads to server for {cooldown:?} (failure #{consecutive_failures}): {}", + match origin { + Origin::Tuple(scheme, host, port) => + format!("{}://{}:{}", scheme, host, port), + Origin::Opaque(_) => "[opaque origin]".to_string(), + } + ); + } } - - // Note: we don't particularly care much for the status code since - // this is just a generic implementation. } debug!("Background uploader loop has exited"); } } +impl BackgroundUploader { + /// Signals the uploader to shut down, waiting up to `deadline` for its + /// background thread to drain whatever is already queued and exit. + /// + /// `Drop`'s own shutdown (used when this isn't called explicitly) waits + /// unboundedly: `origin_cooldowns` (see `start_upload_task`) bounds the + /// wait for an origin that outright *fails*, since further requests to + /// it within its current backoff window are skipped outright, but a + /// slow-but-successful endpoint is not bounded that way -- every queued + /// request still gets its own attempt, each up to that request's own + /// timeout, so draining a large backlog could still take a while. + /// Callers for whom that matters (telemetry in particular: "must never + /// meaningfully delay Trident's actual work" is a stated design goal + /// here) should call this explicitly instead of just letting the value + /// drop. + /// + /// If `deadline` elapses first, the background thread is abandoned + /// (its remaining queued requests may still complete before the + /// process actually exits, but this call returns without waiting + /// further for them). + pub fn shutdown_with_deadline(mut self, deadline: Duration) { + let Some((sender, handle)) = self.inner.take() else { + return; + }; + drop(sender); + + match join_with_deadline(handle, deadline) { + Ok(Ok(())) => debug!("Background uploader shut down"), + Ok(Err(e)) => error!("Background uploader thread panicked: {:?}", e), + Err(_) => { + debug!("Background uploader did not shut down within {deadline:?}; abandoning it") + } + } + } +} + +/// Waits up to `deadline` for `handle` to finish, returning its result if it +/// does. `JoinHandle::join` has no built-in timeout, so this moves the +/// actual join onto a throwaway thread and applies the timeout via a +/// channel receive instead; if `deadline` elapses first, that throwaway +/// thread (and by extension whatever `handle` was waiting on) is +/// abandoned rather than awaited further. +fn join_with_deadline( + handle: JoinHandle, + deadline: Duration, +) -> Result, std::sync::mpsc::RecvTimeoutError> { + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let _ = Builder::new() + .name("background-uploader-shutdown-watcher".into()) + .spawn(move || { + let _ = done_tx.send(handle.join()); + }); + done_rx.recv_timeout(deadline) +} + impl Drop for BackgroundUploader { fn drop(&mut self) { // When the sender is dropped, the upload loop will exit gracefully @@ -141,12 +305,30 @@ pub struct BackgroundUploadHandle { } impl BackgroundUploadHandle { - /// Sends data to be uploaded in the background. + /// Sends data to be uploaded in the background. Any 2xx response is + /// treated as success; use [`Self::upload_with_validator`] if the + /// destination's ingestion protocol can reject part of a request + /// while still returning a 2xx status. pub fn upload( &self, url: &Url, body: impl Into>, timeout: Duration, + content_type: Option<&'static str>, + ) -> Result<(), Error> { + self.upload_with_validator(url, body, timeout, content_type, None) + } + + /// Same as [`Self::upload`], but with an optional response validator + /// -- see `UploadData`'s `response_validator` field doc comment above + /// for its contract. + pub fn upload_with_validator( + &self, + url: &Url, + body: impl Into>, + timeout: Duration, + content_type: Option<&'static str>, + response_validator: Option, ) -> Result<(), Error> { if let Some(sender) = self.sender.upgrade() { sender @@ -154,6 +336,8 @@ impl BackgroundUploadHandle { url: url.clone(), body: body.into(), timeout, + content_type, + response_validator, }) .context("Failed to send data to background uploader") } else { @@ -176,7 +360,7 @@ impl BackgroundUploadHandle { mod tests { use super::*; - use std::time::Duration; + use std::time::{Duration, Instant}; use mockito::{Matcher, Server}; @@ -187,6 +371,19 @@ mod tests { .try_init(); } + #[test] + /// The backoff schedule should start at the base cooldown, double with + /// each consecutive failure, and clamp at the configured maximum + /// instead of growing unbounded (or overflowing) for a long-dead + /// origin that keeps failing indefinitely. + fn test_backoff_for_failures_doubles_and_clamps() { + assert_eq!(backoff_for_failures(1), BASE_ORIGIN_COOLDOWN); + assert_eq!(backoff_for_failures(2), BASE_ORIGIN_COOLDOWN * 2); + assert_eq!(backoff_for_failures(3), BASE_ORIGIN_COOLDOWN * 4); + assert_eq!(backoff_for_failures(100), MAX_ORIGIN_COOLDOWN); + assert_eq!(backoff_for_failures(u32::MAX), MAX_ORIGIN_COOLDOWN); + } + fn run_in_runtime(f: impl std::future::Future) { let runtime = tokio::runtime::Builder::new_current_thread() .enable_all() @@ -208,7 +405,7 @@ mod tests { let url = Url::parse("http://example.invalid/upload").unwrap(); // After shutdown, the weak sender can't be upgraded so upload should error. let err = handle - .upload(&url, b"hello".to_vec(), Duration::from_millis(50)) + .upload(&url, b"hello".to_vec(), Duration::from_millis(50), None) .unwrap_err(); assert!( err.to_string().contains("shut down"), @@ -236,7 +433,7 @@ mod tests { let url = Url::parse(&server.url()).unwrap().join("/upload").unwrap(); handle - .upload(&url, body.as_bytes().to_vec(), Duration::from_secs(2)) + .upload(&url, body.as_bytes().to_vec(), Duration::from_secs(2), None) .unwrap(); // Drop uploader first to ensure the background thread finishes processing all queued @@ -272,6 +469,8 @@ mod tests { url, body: body.as_bytes().to_vec(), timeout: Duration::from_secs(2), + content_type: None, + response_validator: None, }) .unwrap(); @@ -323,6 +522,8 @@ mod tests { url: Url::parse(&server.url()).unwrap().join("/slow").unwrap(), body: b"timeout-me".to_vec(), timeout: Duration::from_millis(100), + content_type: None, + response_validator: None, }) .unwrap(); @@ -332,6 +533,8 @@ mod tests { url: Url::parse(&server.url()).unwrap().join("/upload").unwrap(), body: b"this-should-be-skipped".to_vec(), timeout: Duration::from_secs(2), + content_type: None, + response_validator: None, }) .unwrap(); @@ -347,6 +550,62 @@ mod tests { should_not_hit.assert(); } + #[test] + /// Directly tests that a 3xx response (which `error_for_status()` alone would treat as + /// success) is still handled as an upload failure: the origin gets ignored for later + /// uploads, just like a 4xx/5xx response or a network-level error. + fn test_upload_loop_redirect_status_is_treated_as_failure() { + init_test_logging(); + + let mut server = Server::new(); + let redirect_mock = server + .mock("POST", "/redirect") + .with_status(302) + .expect(1) + .create(); + + let should_not_hit = server + .mock("POST", "/upload") + .with_status(200) + .expect(0) + .create(); + + let (sender, receiver) = mpsc::unbounded_channel::(); + + sender + .send(UploadData { + url: Url::parse(&server.url()) + .unwrap() + .join("/redirect") + .unwrap(), + body: b"redirect-me".to_vec(), + timeout: Duration::from_secs(2), + content_type: None, + response_validator: None, + }) + .unwrap(); + + // Same origin; should be skipped after the first is treated as a failure. + sender + .send(UploadData { + url: Url::parse(&server.url()).unwrap().join("/upload").unwrap(), + body: b"this-should-be-skipped".to_vec(), + timeout: Duration::from_secs(2), + content_type: None, + response_validator: None, + }) + .unwrap(); + + drop(sender); + + run_in_runtime(async { + BackgroundUploader::upload_loop(receiver).await; + }); + + redirect_mock.assert(); + should_not_hit.assert(); + } + #[test] /// Directly tests `upload_loop` shutdown behavior: once the channel is closed, the loop /// should upload remaining items in the queue before exiting. @@ -368,6 +627,8 @@ mod tests { url: Url::parse(&server.url()).unwrap().join("/queued").unwrap(), body: b"queued".to_vec(), timeout: Duration::from_secs(1), + content_type: None, + response_validator: None, }) .unwrap(); // Close the sender before running the loop to simulate shutdown. @@ -402,7 +663,7 @@ mod tests { let url = Url::parse(&server.url()).unwrap().join("/ok").unwrap(); handle - .upload(&url, b"hello".to_vec(), Duration::from_secs(2)) + .upload(&url, b"hello".to_vec(), Duration::from_secs(2), None) .unwrap(); // Drop the uploader to shut down the background thread. Both `handle` @@ -423,9 +684,61 @@ mod tests { &Url::parse(&server.url()).unwrap().join("/nope").unwrap(), b"nope".to_vec(), Duration::from_secs(1), + None, ) .unwrap_err(); assert!(err.to_string().contains("shut down")); after_drop.assert(); } + + #[test] + fn test_shutdown_with_deadline_returns_promptly_with_empty_queue() { + init_test_logging(); + + let uploader = BackgroundUploader::new().unwrap(); + let start = Instant::now(); + uploader.shutdown_with_deadline(Duration::from_secs(5)); + assert!( + start.elapsed() < Duration::from_secs(1), + "shutdown with nothing queued should be immediate" + ); + } + + #[test] + /// Deliberately not exercised via a real `BackgroundUploader` + + /// network mock: a genuinely abandoned background thread would keep + /// running past this test's own scope, in a process shared with every + /// other test in the suite, risking exactly the kind of cross-test + /// port/resource collisions a slow real HTTP mock invites under + /// `cargo test`'s default parallelism. `join_with_deadline` is pure + /// std-only plumbing (a thread + a timed channel receive), so testing + /// it directly with a plain `thread::spawn` gives the same coverage + /// without that risk. + fn test_join_with_deadline_abandons_a_slow_thread() { + let (release_tx, release_rx) = std::sync::mpsc::channel::<()>(); + let handle = std::thread::spawn(move || { + // Blocks until the test explicitly releases it below, standing + // in for a still-busy background uploader thread. + let _ = release_rx.recv(); + }); + + let start = Instant::now(); + let result = join_with_deadline(handle, Duration::from_millis(50)); + let elapsed = start.elapsed(); + + assert!( + result.is_err(), + "join_with_deadline should report a timeout, not a completed join" + ); + assert!( + elapsed < Duration::from_secs(1), + "join_with_deadline should return near its deadline, not block on \ + the still-running thread; took {elapsed:?}" + ); + + // Unlike the real "abandon" scenario this stands in for, we can + // cleanly unblock the spawned thread here, so it exits rather than + // lingering for the rest of the test binary's process lifetime. + let _ = release_tx.send(()); + } } diff --git a/crates/trident/src/logging/logstream.rs b/crates/trident/src/logging/logstream.rs index 8d19a36b2..d46592f09 100644 --- a/crates/trident/src/logging/logstream.rs +++ b/crates/trident/src/logging/logstream.rs @@ -177,7 +177,10 @@ impl Log for LogSender { // Send logs with a reasonably low timeout. The uploader will drop // logs if the server is unreachable or slow, or if it has been // closed. - if let Err(e) = self.uploader.upload(&target, body, Duration::from_secs(5)) { + if let Err(e) = self + .uploader + .upload(&target, body, Duration::from_secs(5), None) + { if !self.send_failed.swap(true, Ordering::Relaxed) { eprintln!("Failed to send log entry: {e}"); } diff --git a/crates/trident/src/logging/mod.rs b/crates/trident/src/logging/mod.rs index 082b1be2c..a94293afd 100644 --- a/crates/trident/src/logging/mod.rs +++ b/crates/trident/src/logging/mod.rs @@ -1,10 +1,12 @@ use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; +pub(super) mod appinsights; pub(super) mod background_log; pub(super) mod background_uploader; pub(super) mod logfwd; pub(super) mod logstream; +pub(super) mod operation_context; pub(super) mod tracestream; #[derive(Debug, Serialize, Deserialize)] diff --git a/crates/trident/src/logging/operation_context.rs b/crates/trident/src/logging/operation_context.rs new file mode 100644 index 000000000..71d2103fc --- /dev/null +++ b/crates/trident/src/logging/operation_context.rs @@ -0,0 +1,370 @@ +//! Thread-local "which command is currently executing, under what +//! operation ID, and from which of Trident's three entry points" context, +//! so telemetry sinks ([`super::tracestream::TraceSender`], +//! [`super::appinsights::AppInsightsSender`]) can tag every metric/span +//! fired during a command's execution with `command`/`operation_id`/ +//! `source` fields, without every call site (deep in `engine::*`, +//! `Trident::*`, etc.) needing to pass them explicitly. +//! +//! A thread-local (rather than e.g. a `tracing` span) is enough here +//! because all three places that set this context run the entire command +//! synchronously on a single, dedicated thread for the command's whole +//! duration: +//! - CLI: `run_trident`'s command dispatch (synchronous, main thread), +//! tagged [`OperationSource::Cli`]. +//! - gRPC/daemon: `servicing_request`'s closure runs inside +//! `tokio::task::spawn_blocking`, which gives it its own OS thread for +//! as long as the closure runs, tagged [`OperationSource::Daemon`]. +//! - gRPC client: `grpc_client`'s command dispatch (synchronous, main +//! thread of the CLI process acting as a client of a running daemon), +//! tagged [`OperationSource::GrpcClient`]. +//! +//! `operation_id` is a fresh, random ID generated once per command +//! invocation (distinct from the persistent, per-host +//! `DataStore::correlation_id`, which is unrelated and set separately on +//! `TraceStream`/`AppInsightsSender`). + +use std::{cell::RefCell, sync::Mutex}; + +use uuid::Uuid; + +/// 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 +/// request it was trying to reach, or from a direct CLI invocation that +/// bypassed the daemon entirely. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OperationSource { + /// A command run directly by the CLI, without going through the + /// daemon (e.g. `trident install` on a host with no daemon running). + Cli, + /// A command executed by the daemon in response to a gRPC request + /// (see `server::tridentserver::TridentServer::servicing_request`). + Daemon, + /// A command run by the CLI acting as a gRPC client, relaying the + /// request to a running daemon (see `grpc_client`). + GrpcClient, +} + +impl OperationSource { + pub fn as_str(&self) -> &'static str { + match self { + OperationSource::Cli => "cli", + OperationSource::Daemon => "daemon", + OperationSource::GrpcClient => "grpc-client", + } + } +} + +thread_local! { + static CURRENT_OPERATION: RefCell> = + const { RefCell::new(None) }; +} + +/// Runs `f` with this thread tagged as executing `command` from `source`, +/// under a fresh `operation_id`. Also fires a `command_start` metric event +/// immediately, tagged the same way. Clears the tag afterwards (even if +/// `f` panics, via a drop guard), so a thread that runs multiple commands +/// over its lifetime (e.g. a thread pool worker reused across +/// `spawn_blocking` calls) never leaks a stale tag into an unrelated +/// later command. +/// +/// The context (and its drop guard) is installed *before* firing +/// `command_start`, and that event carries only `metric_name` -- not +/// explicit `command`/`operation_id`/`source` fields. Both telemetry sinks +/// (`TraceSender`, `AppInsightsSender`) read the just-installed context via +/// `current()` and merge `command`/`operation_id`/`source` into the same +/// `additional_fields`/properties map every other event during this +/// invocation gets them from. Emitting them as explicit fields on +/// `command_start` itself, before the context existed, would instead land +/// them in that event's own `value`/properties body -- a different schema +/// from every other event, and invisible to consumers that only look at +/// `additional_fields` for operation metadata. +pub fn run_with_operation(command: &str, source: OperationSource, f: impl FnOnce() -> R) -> R { + let operation_id = Uuid::new_v4().to_string(); + + CURRENT_OPERATION.with(|cell| { + *cell.borrow_mut() = Some((operation_id, command.to_string(), source)); + }); + + struct ClearOnDrop; + impl Drop for ClearOnDrop { + fn drop(&mut self) { + CURRENT_OPERATION.with(|cell| *cell.borrow_mut() = None); + } + } + let _clear = ClearOnDrop; + + tracing::info!(metric_name = "command_start"); + + f() +} + +/// Returns the `(operation_id, command, source)` triple set by +/// [`run_with_operation`] for the calling thread, if any. +pub(crate) fn current() -> Option<(String, String, OperationSource)> { + CURRENT_OPERATION.with(|cell| cell.borrow().clone()) +} + +/// A snapshot of another thread's operation context (see +/// [`run_with_operation`]), capturable via [`snapshot`] and re-installed +/// on a different thread via [`run_with_captured_operation`]. Used to +/// propagate `operation_id`/`command`/`source` into threads spawned +/// mid-command (e.g. `MonitorMetrics`'s background sampling thread), +/// which otherwise start with no thread-local context of their own and +/// would silently drop these fields from their own metrics. +#[derive(Clone)] +pub struct CapturedOperation(String, String, OperationSource); + +/// Captures the calling thread's current operation context, if any, for +/// later re-installation on another thread via +/// [`run_with_captured_operation`]. Call this on the *spawning* thread, +/// before handing the result to the new thread's closure. +pub fn snapshot() -> Option { + current() + .map(|(operation_id, command, source)| CapturedOperation(operation_id, command, source)) +} + +/// Runs `f` with `captured` (from [`snapshot`]) installed as the calling +/// thread's operation context for the duration of `f`, clearing it +/// afterwards (even on panic). Unlike [`run_with_operation`], this does +/// *not* mint a new `operation_id` or fire a `command_start` metric -- it +/// re-uses an existing operation's identity on a different thread rather +/// than starting a new one. A `None` `captured` (e.g. the spawning thread +/// itself had no operation context -- this thread was started outside any +/// command) makes this a plain, untagged call to `f()`. +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)); + }); + + struct ClearOnDrop; + impl Drop for ClearOnDrop { + fn drop(&mut self) { + CURRENT_OPERATION.with(|cell| *cell.borrow_mut() = None); + } + } + let _clear = ClearOnDrop; + + f() +} + +/// Holds a snapshot of the just-finished servicing operation's context, +/// captured by [`save_reboot_operation`] the moment that operation +/// decides a reboot is needed, and consumed exactly once by +/// [`take_reboot_operation`] at the actual post-servicing reboot call +/// site (CLI: `main.rs`; daemon: `server::reboot`). +/// +/// A process-global (not thread-local) slot is required here, unlike +/// [`snapshot`]/[`run_with_captured_operation`] above: those propagate a +/// context to a thread spawned *while the original context is still +/// active* (the spawning thread hands the snapshot directly to the new +/// thread's closure). Here, by the time the reboot call happens, the +/// operation that decided a reboot was needed has already returned -- +/// its `run_with_operation` scope (and thread-local context) is gone -- +/// and the reboot call itself runs later, on a different thread/call +/// stack that has no way to receive a snapshot as a direct parameter (the +/// CLI's `main` and the daemon's `server_main` reboot path both reach the +/// reboot call through several layers of return values -- `ExitKind`, +/// `Completed`, etc. -- that don't carry operation context). A shared +/// slot lets the operation stash its own context just before it returns, +/// for the reboot call to pick up moments later regardless of which +/// thread it ends up running on. +static PENDING_REBOOT_OPERATION: Mutex> = Mutex::new(None); + +/// Captures the calling thread's current operation context (if any) into +/// [`PENDING_REBOOT_OPERATION`]. Call this from inside the servicing +/// operation's own context, as soon as it decides a reboot is needed -- +/// i.e. while that context is still installed -- so the *original* +/// servicing invocation's `operation_id`/`command` (not a fresh "reboot" +/// identity) can later be attached to `trident_system_reboot` and any +/// telemetry from the reboot call itself, via [`take_reboot_operation`]. +pub fn save_reboot_operation() { + *PENDING_REBOOT_OPERATION.lock().unwrap() = snapshot(); +} + +/// Takes (clearing) whichever operation context was most recently saved +/// via [`save_reboot_operation`], for use with +/// [`run_with_captured_operation`] at the actual reboot call site. `None` +/// if no operation ever called [`save_reboot_operation`] (e.g. reboot +/// requested outside any servicing operation), or if it was already +/// consumed by a prior call. +pub fn take_reboot_operation() -> Option { + PENDING_REBOOT_OPERATION.lock().unwrap().take() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_no_operation_by_default() { + assert!(current().is_none()); + } + + #[test] + fn test_run_with_operation_sets_and_clears_context() { + assert!(current().is_none()); + + let observed = run_with_operation("test_command", OperationSource::Cli, current); + let (operation_id, command, source) = observed.expect("context should be set inside f"); + assert_eq!(command, "test_command"); + assert_eq!(operation_id.len(), 36, "operation_id should be a UUID"); + assert_eq!(source, OperationSource::Cli); + + assert!( + current().is_none(), + "context must be cleared after run_with_operation returns" + ); + } + + #[test] + fn test_run_with_operation_clears_context_on_panic() { + assert!(current().is_none()); + + let result = std::panic::catch_unwind(|| { + run_with_operation("panicking_command", OperationSource::Cli, || { + panic!("boom"); + }) + }); + assert!(result.is_err()); + + assert!( + current().is_none(), + "context must be cleared even if f panics" + ); + } + + #[test] + fn test_each_invocation_gets_a_fresh_operation_id() { + let first = run_with_operation("cmd", OperationSource::Cli, || current().unwrap().0); + let second = run_with_operation("cmd", OperationSource::Cli, || current().unwrap().0); + assert_ne!( + first, second, + "each command invocation gets a fresh operation_id" + ); + } + + #[test] + fn test_snapshot_is_none_outside_run_with_operation() { + assert!(snapshot().is_none()); + } + + #[test] + fn test_snapshot_and_captured_operation_propagates_across_threads() { + // Capture on a thread standing in for the "spawning" thread (here, + // just the current thread inside run_with_operation), then install + // it on a different OS thread, mirroring MonitorMetrics's use. + let (expected_operation_id, expected_command, expected_source, observed) = + run_with_operation( + "cmd_from_parent_thread", + OperationSource::GrpcClient, + || { + let captured = snapshot().expect("should capture a context"); + let (operation_id, command, source) = current().unwrap(); + + let observed = std::thread::spawn(move || { + // No context on a fresh thread until installed. + assert!(current().is_none()); + run_with_captured_operation(Some(captured), current) + }) + .join() + .unwrap(); + + (operation_id, command, source, observed) + }, + ); + + assert_eq!( + observed, + Some((expected_operation_id, expected_command, expected_source)), + "captured operation_id/command/source should propagate to the new thread" + ); + } + + #[test] + fn test_run_with_captured_operation_none_is_a_plain_call() { + assert!(current().is_none()); + let result = run_with_captured_operation(None, current); + assert!(result.is_none()); + assert!(current().is_none()); + } + + #[test] + fn test_run_with_captured_operation_clears_context_after_returning() { + // run_with_captured_operation is meant for a *fresh* thread with no + // context of its own (see MonitorMetrics's use), not nested on top + // of an existing run_with_operation on the *same* thread -- both + // share one flat thread-local slot, so nesting on one thread isn't + // a supported combination. Verify the fresh-thread case clears + // itself after returning. + let captured = run_with_operation("cmd", OperationSource::Daemon, snapshot); + let still_set_inside = std::thread::spawn(move || { + run_with_captured_operation(captured, || current().is_some()) + }) + .join() + .unwrap(); + assert!(still_set_inside, "context should be set while f runs"); + } + + /// Serializes the reboot-operation tests below: unlike the + /// thread-local `CURRENT_OPERATION`, `PENDING_REBOOT_OPERATION` is a + /// process-wide slot, so tests touching it would otherwise race + /// against each other under cargo's default parallel test execution. + static REBOOT_OPERATION_TEST_LOCK: Mutex<()> = Mutex::new(()); + + #[test] + fn test_take_reboot_operation_is_none_by_default() { + let _guard = REBOOT_OPERATION_TEST_LOCK.lock().unwrap(); + assert!(take_reboot_operation().is_none()); + } + + #[test] + fn test_save_and_take_reboot_operation_round_trips() { + 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 captured = take_reboot_operation().expect("should have captured a context"); + let observed = run_with_captured_operation(Some(captured), current).unwrap(); + assert_eq!( + observed, expected, + "the original install's operation_id/command should have been captured" + ); + + assert!( + take_reboot_operation().is_none(), + "take_reboot_operation should clear the slot after taking it" + ); + } + + #[test] + fn test_save_reboot_operation_outside_run_with_operation_is_none() { + let _guard = REBOOT_OPERATION_TEST_LOCK.lock().unwrap(); + assert!( + take_reboot_operation().is_none(), + "start with a clean slate" + ); + + save_reboot_operation(); + + assert!( + take_reboot_operation().is_none(), + "nothing to capture outside an active operation" + ); + } +} diff --git a/crates/trident/src/logging/tracestream.rs b/crates/trident/src/logging/tracestream.rs index d6f87fdf0..531128f49 100644 --- a/crates/trident/src/logging/tracestream.rs +++ b/crates/trident/src/logging/tracestream.rs @@ -2,6 +2,7 @@ use std::{ collections::BTreeMap, fs::{self, File}, io::Write, + path::Path, sync::{Arc, RwLock}, time::Instant, }; @@ -18,13 +19,17 @@ use tracing::{ }; use tracing_subscriber::{layer::Layer, registry::LookupSpan}; +use trident_api::error::TridentError; + use osutils::{ files, osrelease::{OsRelease, OS_RELEASE_PATH}, uname, }; -use crate::{TRIDENT_METRICS_FILE_PATH, TRIDENT_VERSION}; +use crate::{ + datastore::DataStore, logging::operation_context, TRIDENT_METRICS_FILE_PATH, TRIDENT_VERSION, +}; /// The product uuid is used to identify the hardware that Trident is running on. const PRODUCT_UUID_FILE: &str = "/sys/class/dmi/id/product_uuid"; @@ -84,6 +89,10 @@ pub struct TraceStream { // TODO: Consider changing this to a LockOnce when rustc is updated to // >=1.70 target: Arc>>, + installation_id: Arc>>, + /// Stable for the lifetime of the datastore, unlike `installation_id` + /// which is tied to a specific `Trident::install` invocation. See + /// `crate::datastore::DataStore::datastore_id`. datastore_id: Arc>>, disabled: bool, } @@ -126,11 +135,44 @@ impl TraceStream { Ok(()) } - /// Set the database ID to attach to every trace entry sent from this point - /// forward, as an additional field, so that all traces/metrics for a - /// given host installation can be correlated. Expected to be called once - /// the datastore's persisted database ID has been retrieved (see - /// `DataStore::datastore_id`). + /// Set the installation ID to attach to every trace entry sent from this + /// point forward, as an additional field, so that all traces/metrics for + /// a given host installation can be correlated. Expected to be called + /// once the datastore's persisted installation ID has been retrieved + /// (see [`Self::attach_installation_id_if_present`] and + /// [`Self::create_and_attach_installation_id`]). + pub fn set_installation_id(&self, installation_id: String) { + match self.installation_id.write() { + Ok(mut val) => { + val.replace(installation_id); + } + Err(_) => warn!("Failed to lock tracestream to set installation ID"), + } + } + + /// Returns a clone of the shared installation-ID handle -- the same + /// underlying `Arc>` written by `set_installation_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 installation_id_handle(&self) -> Arc>> { + self.installation_id.clone() + } + + fn installation_id_cached(&self) -> bool { + self.installation_id + .read() + .map(|v| v.is_some()) + .unwrap_or(false) + } + + /// Set the database ID to attach to every trace entry sent from this + /// point forward, as an additional field, so that all traces/metrics + /// against a given datastore can be correlated. Unlike + /// `installation_id`, this is stable for the datastore's entire + /// lifetime, not just a single `Trident::install` invocation. Expected + /// to be called once the datastore's persisted database ID has been + /// retrieved (see [`Self::attach_datastore_id_if_present`]). pub fn set_datastore_id(&self, datastore_id: String) { match self.datastore_id.write() { Ok(mut val) => { @@ -140,6 +182,112 @@ impl TraceStream { } } + /// Returns a clone of the shared database-ID handle -- the same + /// underlying `Arc>` written by `set_datastore_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 datastore_id_handle(&self) -> Arc>> { + self.datastore_id.clone() + } + + fn datastore_id_cached(&self) -> bool { + self.datastore_id + .read() + .map(|v| v.is_some()) + .unwrap_or(false) + } + + /// Best-effort attempt to attach this datastore's database ID -- either + /// 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: unlike `installation_id`, the + /// database ID's get-or-create semantics only ever run against a + /// datastore that has already been opened for real (see + /// `crate::datastore::DataStore::datastore_id`), so it is safe to call + /// this any time a datastore is known to already exist. + pub fn attach_datastore_id_if_present(&self, datastore_path: &Path) { + if self.datastore_id_cached() || !datastore_path.exists() { + return; + } + match DataStore::open(datastore_path).and_then(|mut ds| ds.datastore_id()) { + Ok(datastore_id) => { + info!("Datastore ID: {datastore_id}"); + self.set_datastore_id(datastore_id.to_string()); + } + Err(e) => { + warn!("Failed to read/create database ID: {e:?}"); + } + } + } + + /// Best-effort attempt to attach this host's installation ID -- either + /// 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::create_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. + /// + /// Safe to call from anywhere, any number of times, before any point + /// that wants the 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, the daemon's + /// per-request backstop, and `Trident::new`'s own attach), so a + /// correctness fix to this logic only needs to happen once. + pub fn attach_installation_id_if_present(&self, datastore_path: &Path) { + if self.installation_id_cached() || !datastore_path.exists() { + return; + } + match DataStore::open(datastore_path).and_then(|mut ds| ds.installation_id_or_migrate()) { + Ok(Some(installation_id)) => { + info!("Installation ID: {installation_id}"); + self.set_installation_id(installation_id.to_string()); + } + Ok(None) => { + debug!("No installation ID persisted yet (host not yet installed)"); + } + Err(e) => { + warn!("Failed to read installation ID: {e:?}"); + } + } + } + + /// Creates (or reads back, if one already exists) `datastore`'s + /// installation ID and attaches it. Unlike + /// [`Self::attach_installation_id_if_present`], this is only for the + /// one caller that already knows a command genuinely allowed to + /// initialize a brand-new datastore (per + /// [`crate::datastore::DataStore::may_initialize_datastore_for_command`]) + /// is proceeding, and already holds (or just created) the datastore + /// handle for it -- so this attaches the new install/update's own ID + /// instead of leaving the trace stream untagged until some later + /// read-only attach happens to run. + pub fn create_and_attach_installation_id( + &self, + datastore: &mut DataStore, + ) -> Result<(), TridentError> { + if self.installation_id_cached() { + return Ok(()); + } + let installation_id = datastore.create_installation_id()?; + info!("Installation ID: {installation_id}"); + self.set_installation_id(installation_id.to_string()); + Ok(()) + } + /// Create a Boxed TraceSender pub fn make_trace_sender(&self) -> Box { self.make_trace_sender_with_metrics_path(TRIDENT_METRICS_FILE_PATH) @@ -157,6 +305,7 @@ impl TraceStream { ) -> Box { Box::new(TraceSender::new( self.target.clone(), + self.installation_id.clone(), self.datastore_id.clone(), metrics_file_path, )) @@ -165,6 +314,7 @@ impl TraceStream { pub struct TraceSender { server: Arc>>, + installation_id: Arc>>, datastore_id: Arc>>, client: reqwest::blocking::Client, metrics_file: Option, @@ -179,11 +329,13 @@ struct ExecutionTime(Instant); impl TraceSender { fn new( server: Arc>>, + installation_id: Arc>>, datastore_id: Arc>>, metrics_file_path: &str, ) -> Self { Self { server, + installation_id, datastore_id, client: reqwest::blocking::Client::new(), metrics_file: match files::create_file(metrics_file_path) { @@ -203,16 +355,42 @@ impl TraceSender { } /// Build the `additional_fields` map for a trace entry: the static - /// `ADDITIONAL_FIELDS`, plus the database ID (if one has been set via - /// `TraceStream::set_datastore_id`), so entries can be correlated back to a - /// specific host installation. + /// `ADDITIONAL_FIELDS`, the installation ID (if one has been set via + /// `TraceStream::set_installation_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. + /// + /// `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 filled in by two different paths: normally + /// from the persisted value set via `TraceStream::set_installation_id` + /// (attached above), but `merge_operation_context` also falls back to + /// this invocation's own `operation_id` whenever no persisted value + /// has been attached yet -- e.g. every event fired before a host's + /// first-ever `install` has actually created the datastore and created + /// one. See `merge_operation_context` for why that fallback is the + /// same value `create_installation_id` will end up persisting for + /// that same invocation. fn additional_fields(&self) -> 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("installation_id".to_string(), json!(installation_id)); + } + } if let Ok(datastore_id) = self.datastore_id.read() { if let Some(datastore_id) = datastore_id.as_ref() { fields.insert("datastore_id".to_string(), json!(datastore_id)); } } + merge_operation_context(&mut fields); fields } @@ -401,6 +579,40 @@ where } } +/// 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. +/// +/// Shared by both local telemetry sinks (`TraceSender::additional_fields` +/// below and `AppInsightsSender::send_event`) so the +/// operation_id/command/source/installation_id-fallback rule has one +/// implementation instead of being hand-duplicated between them -- a +/// 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) { + if let Some((operation_id, command, source)) = operation_context::current() { + fields + .entry("operation_id".to_string()) + .or_insert_with(|| json!(operation_id)); + fields + .entry("command".to_string()) + .or_insert_with(|| json!(command)); + fields + .entry("source".to_string()) + .or_insert_with(|| json!(source.as_str())); + // If no installation ID has been persisted/attached yet (e.g. this + // is the invocation that is about to create the datastore and + // create one), fall back to this invocation's own `operation_id` -- + // the same value `DataStore::create_installation_id` will persist + // as the installation ID once the datastore is actually created. + fields + .entry("installation_id".to_string()) + .or_insert_with(|| json!(operation_id)); + } +} + /// Obtain product uuid of the hardware Trident is running on fn read_product_uuid(filepath: String) -> String { match fs::read_to_string(filepath.clone()) { @@ -579,11 +791,53 @@ mod tests { } #[test] - /// Regression test: `TraceStream::set_datastore_id` must actually - /// reach the serialized trace entry's `additional_fields.datastore_id` + /// Regression test: `TraceStream::set_installation_id` must actually + /// reach the serialized trace entry's `additional_fields.installation_id` /// -- the metric/span tests above only assert on `metric_name`/`value` - /// and would still pass even if the database ID were never copied + /// and would still pass even if the installation ID were never copied /// into `additional_fields`. + fn test_tracestream_installation_id_written_to_additional_fields() { + let temp_dir = tempfile::tempdir().unwrap(); + let metrics_path = temp_dir.path().join("metrics.jsonl"); + 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()) + .with_filter(filter::LevelFilter::INFO); + + // See test_tracestream_write_metric_event_to_file for why a scoped + // (not global) default subscriber is used here. + let _guard = tracing::subscriber::set_default( + tracing_subscriber::Registry::default().with(trace_sender), + ); + + tracing::info!( + metric_name = "test_metric_with_installation_id", + value = true + ); + + // Ensure the trace system has time to write the file. + 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_with_installation_id""#) + && line.contains(r#""installation_id":"test-installation-id""#) + }); + + assert!( + metric_found, + "Expected metric with installation_id field not found in the local metrics file" + ); + } + + #[test] + /// Regression test: `TraceStream::set_datastore_id` must actually + /// reach the serialized trace entry's `additional_fields.datastore_id`, + /// independently of installation_id. fn test_tracestream_datastore_id_written_to_additional_fields() { let temp_dir = tempfile::tempdir().unwrap(); let metrics_path = temp_dir.path().join("metrics.jsonl"); diff --git a/crates/trident/src/main.rs b/crates/trident/src/main.rs index c1d996459..132c64480 100644 --- a/crates/trident/src/main.rs +++ b/crates/trident/src/main.rs @@ -1,8 +1,8 @@ -use std::{fs, iter, panic, process::ExitCode}; +use std::{fs, iter, panic, process::ExitCode, time::Duration}; use anyhow::{Context, Error}; use clap::Parser; -use log::{error, info, LevelFilter, Log}; +use log::{error, info, warn, LevelFilter, Log}; use osutils::logging::{filter::LogFilter, multilog::MultiLogger}; use trident::{ @@ -10,14 +10,34 @@ use trident::{ cli::{self, Cli, Commands, GetKind, TridentExitCodes}, init::offline, manual_rollback::{self, utils::ManualRollbackRequestKind}, - validation, BackgroundLog, BackgroundUploader, DataStore, ExitKind, LogForwarder, Logstream, - TraceStream, Trident, TRIDENT_BACKGROUND_LOG_PATH, + 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, }; use trident_api::{ - config::HostConfigurationSource, + config::{HostConfigurationSource, Operations}, 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, @@ -128,6 +148,33 @@ fn run_trident( | 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()) @@ -135,107 +182,181 @@ fn run_trident( Commands::RebuildRaid { config, .. } => config.clone(), _ => None, }; - if let Some(path) = &config_path { if !path.exists() { - return Err(TridentError::new(InvalidInputError::ReadInputFile { - path: path.to_string_lossy().to_string(), - })) - .message("Config file does not exist"); + 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") + }); } } - let agent_config = AgentConfig::load()?; - // For non-install and non-update (update will check and has special handling for CIH - // scenario) commands, we expect the datastore to exist - if !matches!( - args.command, - Commands::Install { .. } | Commands::Update { .. } - ) && !agent_config.datastore_path().exists() - { - return Err(TridentError::new(InvalidInputError::HostNotProvisioned)) - .message("Datastore file does not exist"); + // 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()); } - let mut trident = Trident::new( - config_path.map(HostConfigurationSource::File), - agent_config.datastore_path(), - logstream, - tracestream, - ) - .message("Failed to initialize Trident")?; - - // `Trident::new` has already retrieved (or created) this - // host's persisted database ID and attached it to the - // shared TraceStream, so every trace/metric emitted from - // here on -- including "trident_start" -- carries it. - let mut datastore = DataStore::open_or_create(agent_config.datastore_path()) - .message("Failed to open datastore")?; - - // Execute the command - let res = match args.command { - Commands::Install { - ref allowed_operations, - multiboot, - .. - } => trident - .install( - &mut datastore, - cli::to_operations(allowed_operations), - multiboot, - None, + run_with_operation(&command, OperationSource::Cli, || { + // config_path was already validated (existence-checked) + // above. Reuse the same `AgentConfig` snapshot loaded + // just above rather than reloading -- see the comment + // there. + let agent_config = agent_config_result?; + // For commands that cannot themselves stage a new + // install/update (see + // `DataStore::may_initialize_datastore_for_command`), + // we expect the datastore to already exist. Update has + // its own special handling for the CIH bootstrap + // scenario further down. + if !DataStore::may_initialize_datastore_for_command(&command) + && !agent_config.datastore_path().exists() + { + return Err(TridentError::new(InvalidInputError::HostNotProvisioned)) + .message("Datastore file does not exist"); + } + + // A multiboot install may swap to a brand-new temporary + // datastore inside `Trident::install` (see there), + // distinct from `datastore_path` here (the existing + // host's persistent datastore) -- so defer attaching an + // installation ID until `install` has settled on which + // datastore it actually uses, rather than attaching the + // existing host's here and having it be wrong for the + // rest of the run. See `new_deferring_installation_id`'s + // doc comment for the full rationale. + let defer_installation_id = matches!( + args.command, + Commands::Install { + multiboot: true, + .. + } + ); + let mut trident = if defer_installation_id { + Trident::new_deferring_installation_id( + config_path.map(HostConfigurationSource::File), + agent_config.datastore_path(), + logstream.clone(), + tracestream.clone(), ) - .map(|(exit_kind, _image_hash, _servicing_type)| exit_kind), - Commands::Update { - ref allowed_operations, - .. - } => trident - .update(&mut datastore, cli::to_operations(allowed_operations)) - .map(|(exit_kind, _image_hash, _servicing_type)| exit_kind), - Commands::Commit { .. } => trident - .commit(&mut datastore) - .map(|(exit_kind, _servicing_type)| exit_kind), - Commands::Rollback { - runtime, - ab, - ref allowed_operations, - .. - } => trident - .rollback( - &mut datastore, + } else { + Trident::new( + config_path.map(HostConfigurationSource::File), + agent_config.datastore_path(), + logstream.clone(), + tracestream.clone(), + ) + } + .message("Failed to initialize Trident")?; + + // `Trident::new` (or `Trident::new_deferring_installation_id` + // for a multiboot install) has already attached this + // host's persisted installation ID -- if any -- to the + // shared TraceStream, so every trace/metric emitted + // from here on -- including "trident_start" -- carries + // it once available. + let mut datastore = DataStore::open_or_create(agent_config.datastore_path()) + .message("Failed to open datastore")?; + + // Execute the command + let res = match args.command { + Commands::Install { + ref allowed_operations, + multiboot, + .. + } => { + let ops = cli::to_operations(allowed_operations); + trident + .install(&mut datastore, ops, multiboot, None) + .map(|(exit_kind, _image_hash, _servicing_type)| exit_kind) + } + Commands::Update { + ref allowed_operations, + .. + } => { + let ops = cli::to_operations(allowed_operations); + trident + .update(&mut datastore, ops) + .map(|(exit_kind, _image_hash, _servicing_type)| exit_kind) + } + Commands::Commit { .. } => trident + .commit(&mut datastore) + .map(|(exit_kind, _servicing_type)| exit_kind), + Commands::Rollback { runtime, ab, - cli::to_operations(allowed_operations), - ) - .map(|(exit_kind, _servicing_type)| exit_kind), - Commands::RebuildRaid { .. } => trident - .rebuild_raid(&mut datastore) - .map(|()| ExitKind::Done), - _ => Err(TridentError::internal("Invalid command")), - }; - - // Return Host Status if requested - if status.is_some() { - if let Err(e) = - Trident::get(agent_config.datastore_path(), status, GetKind::Status) - .message("Failed to retrieve Host Status") - { - error!("{e:?}"); + ref allowed_operations, + .. + } => { + let ops = cli::to_operations(allowed_operations); + trident + .rollback(&mut datastore, runtime, ab, ops) + .map(|(exit_kind, _servicing_type)| exit_kind) + } + Commands::RebuildRaid { .. } => trident + .rebuild_raid(&mut datastore) + .map(|()| ExitKind::Done), + _ => Err(TridentError::internal("Invalid command")), + }; + + // Return Host Status if requested + if status.is_some() { + if let Err(e) = + Trident::get(agent_config.datastore_path(), status, GetKind::Status) + .message("Failed to retrieve Host Status") + { + error!("{e:?}"); + } } - } - // Return error if requested - if let Some(error_path) = error.as_ref() { - if let Err(e) = &res { - if let Err(e2) = - fs::write(error_path, serde_yaml::to_string(&e).unwrap_or("".into())) - { - error!("Failed to write error to file: {e2}"); + // Return error if requested + if let Some(error_path) = error.as_ref() { + if let Err(e) = &res { + if let Err(e2) = fs::write( + error_path, + serde_yaml::to_string(&e).unwrap_or("".into()), + ) { + error!("Failed to write error to file: {e2}"); + } } } - } - res.message(format!("Failed to execute '{}' command", args.command)) + // Capture this operation's identity while its context + // is still installed (this closure runs entirely + // inside `run_with_operation`'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 + // `save_reboot_operation`/`take_reboot_operation`. + if matches!(res, Ok(ExitKind::NeedsReboot)) { + save_reboot_operation(); + } + + res.message(format!("Failed to execute '{}' command", args.command)) + }) } _ => unreachable!(), } @@ -305,10 +426,80 @@ fn setup_logging( Ok(logstream) } -fn setup_tracing(args: &Cli) -> Result { - use tracing_subscriber::{filter, layer::SubscriberExt, Layer}; +/// Whether the Application Insights tracing layer ended up active on this +/// invocation, and why not when it didn't. Computed by [`setup_tracing`] and +/// surfaced via [`TelemetryStatus::log`] once real logging is available, so +/// operators can tell -- from the logs alone, without reading source -- +/// whether telemetry should be expected to actually reach Application +/// Insights, rather than silently assuming it based on the `Telemetry=` +/// setting alone (a bad/unreachable connection string, for example, fails +/// silently otherwise). +#[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. + NotApplicable, + /// `Telemetry=OptOut` (the default): telemetry was never attempted. + OptedOut, + /// Opted in, but no usable `AZURE_MONITOR_CONNECTION_STRING` was + /// compiled into this binary at build time (missing, empty, or failed + /// to parse). + NoConnectionString, + /// Opted in with a connection string, but the dedicated telemetry + /// background uploader is unavailable (failed to start, or its handle + /// was already closed). + UploaderUnavailable, + /// Opted in, connection string valid, uploader available: telemetry is + /// actively being sent. + Enabled, +} + +impl TelemetryStatus { + /// Log this status through the real logging pipeline. Must only be + /// called after logging has been initialized (`setup_logging`) -- + /// calling it earlier would silently no-op, since the `log` facade + /// drops everything until a logger is registered. + fn log(self) { + match self { + TelemetryStatus::NotApplicable => {} + TelemetryStatus::OptedOut => { + info!( + "Telemetry: disabled (Telemetry=OptOut, the default, in agent configuration)" + ); + } + TelemetryStatus::NoConnectionString => { + info!( + "Telemetry: opted in, but no usable Application Insights connection string \ + was compiled into this binary -- telemetry is a no-op" + ); + } + TelemetryStatus::UploaderUnavailable => { + warn!( + "Telemetry: opted in, but the telemetry background uploader is \ + unavailable -- telemetry is a no-op" + ); + } + TelemetryStatus::Enabled => { + info!("Telemetry: enabled, sending tracing data to Application Insights"); + } + } + } +} + +fn setup_tracing( + args: &Cli, + telemetry_enabled: bool, + // Dedicated to Application Insights telemetry -- deliberately *not* the + // same `BackgroundUploader` instance used for log forwarding (see + // `main`), so a slow-but-successful telemetry endpoint can never build a + // backlog that delays real log uploads. `None` if telemetry is disabled + // or its uploader failed to start; either way telemetry becomes a no-op. + telemetry_uploader: Option<&BackgroundUploader>, +) -> Result<(TraceStream, TelemetryStatus), Error> { + use tracing_subscriber::{filter, layer::SubscriberExt, Layer, Registry}; let tracestream = TraceStream::default(); + let mut telemetry_status = TelemetryStatus::NotApplicable; match &args.command { Commands::Commit { .. } @@ -318,36 +509,92 @@ fn setup_tracing(args: &Cli) -> Result { | Commands::RebuildRaid { .. } | Commands::Rollback { check: false, .. } | Commands::Update { .. } => { + let mut layers: Vec + Send + Sync>> = vec![Box::new( + tracestream + .make_trace_sender() + .with_filter(filter::LevelFilter::INFO), + )]; + // As functionality moves to the Daemon, move the journald layer to // only be enabled for the Daemon command. Until then, keep it enabled // for all commands to ensure we have tracing info in journald for all // commands. - let baseline_tracing = tracing_subscriber::Registry::default().with( - tracestream - .make_trace_sender() - .with_filter(filter::LevelFilter::INFO), - ); - if let Ok(journald_layer) = tracing_journald::layer() { - tracing::subscriber::set_global_default( - baseline_tracing.with( + match tracing_journald::layer() { + Ok(journald_layer) => { + layers.push(Box::new( journald_layer .with_syslog_identifier("trident-tracing".to_string()) .with_filter(filter::LevelFilter::INFO), - ), - ) - .context("Failed to set global default subscriber")?; - } else { - eprintln!("Failed to connect to journald, falling back to tracing without journald support"); - tracing::subscriber::set_global_default(baseline_tracing) - .context("Failed to set global default subscriber")?; + )); + } + Err(_) => { + eprintln!("Failed to connect to journald, falling back to tracing without journald support"); + } } + + // Best-effort Application Insights telemetry: only added when the + // user has opted in via the Agent Configuration file *and* a + // connection string was compiled into this binary at build time. + // Never fails startup: an empty/unparsable connection string just + // means telemetry stays a no-op. `telemetry_status` records which + // of these applied so the caller can log it once real logging is + // available (see `TelemetryStatus::log`). + telemetry_status = if !telemetry_enabled { + TelemetryStatus::OptedOut + } else { + // A missing/closed uploader (e.g. its background thread + // failed to start) just means telemetry stays a no-op; it + // must never block or fail the rest of tracing setup. + match telemetry_uploader.and_then(|u| u.get_handle()) { + Some(handle) => match AppInsightsSender::from_connection_string( + trident::AZURE_MONITOR_CONNECTION_STRING, + handle, + tracestream.installation_id_handle(), + tracestream.datastore_id_handle(), + ) { + Some(sender) => { + layers.push(Box::new(sender.with_filter(filter::LevelFilter::INFO))); + TelemetryStatus::Enabled + } + None => TelemetryStatus::NoConnectionString, + }, + None => TelemetryStatus::UploaderUnavailable, + } + }; + + tracing::subscriber::set_global_default(Registry::default().with(layers)) + .context("Failed to set global default subscriber")?; } _ => { // no op } } - Ok(tracestream) + Ok((tracestream, telemetry_status)) +} + +/// How long to wait for the dedicated telemetry uploader to drain and +/// shut down before abandoning it (see +/// `BackgroundUploader::shutdown_with_deadline`). Telemetry must never +/// meaningfully delay Trident's actual work, including at shutdown -- a +/// slow-but-successful Application Insights endpoint could otherwise +/// stall process exit for as long as it takes to drain every queued +/// event. +const TELEMETRY_SHUTDOWN_DEADLINE: Duration = Duration::from_secs(5); + +/// Wraps a `BackgroundUploader` so it is always shut down with a bounded +/// deadline when dropped, regardless of which of `main`'s many return +/// points is taken -- `BackgroundUploader`'s own `Drop` impl (used +/// elsewhere, e.g. for `bg_uploader`, which carries real log delivery and +/// is expected to drain fully) waits unboundedly instead. +struct TelemetryUploaderGuard(Option); + +impl Drop for TelemetryUploaderGuard { + fn drop(&mut self) { + if let Some(uploader) = self.0.take() { + uploader.shutdown_with_deadline(TELEMETRY_SHUTDOWN_DEADLINE); + } + } } fn main() -> ExitCode { @@ -363,13 +610,45 @@ fn main() -> ExitCode { } }; + // Whether best-effort Application Insights telemetry is enabled. Loaded + // early (before logging/tracing is set up) since the decision feeds + // directly into setup_tracing(). AgentConfig::load() never actually + // errors today, but default to disabled (OptOut) defensively if that + // ever changes. + let telemetry_enabled = AgentConfig::load() + .map(|config| config.telemetry_enabled()) + .unwrap_or(false); + + // Application Insights telemetry gets its own dedicated uploader/queue, + // entirely separate from `bg_uploader` (which carries real log + // forwarding). Both uploaders drain their queue sequentially on a single + // background thread, so sharing one between telemetry and logs would let + // a slow-but-successful telemetry endpoint build a backlog that delays + // operational log uploads. Failure to start is not fatal: telemetry + // simply becomes a no-op, mirroring failure handling on the handle + // itself. + let telemetry_uploader = telemetry_enabled + .then(|| match BackgroundUploader::new() { + Ok(uploader) => Some(uploader), + Err(e) => { + eprintln!("Failed to initialize telemetry uploader, disabling telemetry: {e:?}"); + None + } + }) + .flatten(); + // Wrapped immediately so every return path in main() below shuts it + // down with a bounded deadline, not BackgroundUploader's own unbounded + // Drop. + let telemetry_uploader = TelemetryUploaderGuard(telemetry_uploader); + // Initialize the telemetry flow - let tracestream = setup_tracing(&args); - if let Err(e) = tracestream { + let tracing_setup = setup_tracing(&args, telemetry_enabled, telemetry_uploader.0.as_ref()); + if let Err(e) = tracing_setup { // Defer to stderr since logging is not yet initialized. eprintln!("Failed to initialize tracing: {e:?}"); return TridentExitCodes::SetupFailed.into(); } + let (tracestream, telemetry_status) = tracing_setup.unwrap(); if let Commands::Daemon { inactivity_timeout, @@ -395,13 +674,14 @@ fn main() -> ExitCode { // Log version on startup info!("Trident version: {}", trident::TRIDENT_VERSION); + telemetry_status.log(); trident::server_main( log_forwarder, *inactivity_timeout, socket_path, logstream.unwrap(), - tracestream.unwrap(), + tracestream, ) } else if let Commands::GrpcClient(client_args) = &args.command { let logstream = setup_logging(&args, &bg_uploader, iter::empty()); @@ -414,6 +694,8 @@ fn main() -> ExitCode { error!("Failed to initialize logstream from environment: {e:?}"); } + telemetry_status.log(); + // Run the client command trident::client_main(client_args) } else { @@ -424,15 +706,31 @@ fn main() -> ExitCode { return TridentExitCodes::SetupFailed.into(); } + telemetry_status.log(); + // Invoke Trident - match run_trident(logstream.unwrap(), tracestream.unwrap(), &args) { + match run_trident(logstream.unwrap(), tracestream, &args) { Ok(ExitKind::Done) => {} Err(e) => { error!("{e:?}"); return TridentExitCodes::Failed.into(); } Ok(ExitKind::NeedsReboot) => { - if let Err(e) = trident::request_reboot_with_wait() { + // 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, + ) { error!("Failed to reboot: {e:?}"); return TridentExitCodes::RebootUnsuccessful.into(); } diff --git a/crates/trident/src/monitor_metrics.rs b/crates/trident/src/monitor_metrics.rs index 275a61636..da35b00ac 100644 --- a/crates/trident/src/monitor_metrics.rs +++ b/crates/trident/src/monitor_metrics.rs @@ -11,6 +11,8 @@ use std::{ time::Duration, }; +use crate::logging::operation_context; + // This constant defines the interval at which the monitoring thread will check for updates. const MONITORING_INTERVAL_MS: u64 = 100; // in milliseconds @@ -232,37 +234,51 @@ impl MonitorMetrics { let mut memory_stat = MemoryStat::new(phase.clone()); let mut network_stat = NetworkStat::new(phase.clone(), init_net_stats); + // Captured on the spawning thread (which is running inside + // run_with_operation/run_command) and re-installed on the new + // monitoring thread below, so its own summary_trace() events at + // the end still carry operation_id/command/servicing_id -- a + // freshly spawned thread otherwise starts with no thread-local + // operation context of its own. + let captured_operation = operation_context::snapshot(); + let join_handle = thread::spawn(move || { - loop { - // Update CPU and memory statistics - if let Ok(process) = Process::myself() { - if let Ok(stat) = process.stat() { - cpu_stat.update((stat.utime + stat.stime) as f64); - memory_stat.update(stat.rss); + operation_context::run_with_captured_operation(captured_operation, || { + loop { + // Update CPU and memory statistics + if let Ok(process) = Process::myself() { + if let Ok(stat) = process.stat() { + cpu_stat.update((stat.utime + stat.stime) as f64); + memory_stat.update(stat.rss); + } } - } - // Update network statistics - if let Ok(dev_stats) = procfs::net::dev_status() { - let stats: Vec<_> = dev_stats.values().collect(); - for stat in stats { - network_stat.update(stat.name.clone(), stat.recv_bytes, stat.sent_bytes); + // Update network statistics + if let Ok(dev_stats) = procfs::net::dev_status() { + let stats: Vec<_> = dev_stats.values().collect(); + for stat in stats { + network_stat.update( + stat.name.clone(), + stat.recv_bytes, + stat.sent_bytes, + ); + } } - } - local_metric_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + local_metric_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst); - if local_stop.load(std::sync::atomic::Ordering::SeqCst) { - break; - } + if local_stop.load(std::sync::atomic::Ordering::SeqCst) { + break; + } - // Sleep for the polling interval - thread::sleep(polling_interval); - } - // Perform summary trace for CPU, memory, and network metrics - // after the monitoring thread is stopped. - cpu_stat.summary_trace(); - memory_stat.summary_trace(); - network_stat.summary_trace(); + // Sleep for the polling interval + thread::sleep(polling_interval); + } + // Perform summary trace for CPU, memory, and network metrics + // after the monitoring thread is stopped. + cpu_stat.summary_trace(); + memory_stat.summary_trace(); + network_stat.summary_trace(); + }); }); self.join_handle = Some(join_handle); diff --git a/crates/trident/src/server/mod.rs b/crates/trident/src/server/mod.rs index fb8ce91ec..8e7b01dbd 100644 --- a/crates/trident/src/server/mod.rs +++ b/crates/trident/src/server/mod.rs @@ -35,7 +35,7 @@ use trident_proto::v1preview::{ use crate::{ agentconfig::AgentConfig, cli::TridentExitCodes, - logging::logfwd::LogForwarder, + logging::{logfwd::LogForwarder, operation_context}, reboot::{self, REBOOT_WAIT_DURATION_SECS}, ExitKind, Logstream, TraceStream, }; @@ -107,6 +107,21 @@ pub fn server_main( } }; + // Attach this host's installation ID and database ID to the shared + // TraceStream before accepting any RPCs. `Trident::new()` attaches + // both the same way on every request, but the very first servicing + // request this daemon process ever handles would otherwise have its + // command_start (fired by run_with_operation before that request's own + // Trident::new() call runs) go out untagged, so attach them up front + // instead. Every later request is unaffected either way, since the + // shared TraceStream keeps whatever was set here (or by the first + // request) for the rest of the daemon's lifetime. 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`). + tracestream.attach_installation_id_if_present(agent_config.datastore_path()); + tracestream.attach_datastore_id_if_present(agent_config.datastore_path()); + let shutdown_signals = match ShutdownSignals::setup_signal_handlers() { Ok(signals) => signals, Err(e) => { @@ -150,7 +165,20 @@ pub fn server_main( } fn reboot(signals: ShutdownSignals) -> ExitCode { - if let Err(e) = reboot::request_reboot() { + // Reuse the just-finished servicing operation's own operation_id/ + // command (captured via `save_reboot_operation` from inside + // `servicing_request`'s closure, before that operation's own + // `run_with_operation` scope ended) so `trident_system_reboot` -- + // fired unconditionally by `request_reboot` -- is tagged with the + // originating operation instead of left completely untagged: by the + // 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(); } diff --git a/crates/trident/src/server/tridentserver/mod.rs b/crates/trident/src/server/tridentserver/mod.rs index 51434a3db..d3a60a4c2 100644 --- a/crates/trident/src/server/tridentserver/mod.rs +++ b/crates/trident/src/server/tridentserver/mod.rs @@ -25,7 +25,8 @@ use trident_proto::v1::{ use crate::{ agentconfig::AgentConfig, - logging::logfwd::LogForwarder, + datastore::DataStore, + logging::{logfwd::LogForwarder, operation_context}, server::{activitytracker::ActivityTracker, support::stream::StreamWithLock}, ExitKind, Logstream, TraceStream, }; @@ -177,6 +178,31 @@ impl TridentServer { }) } + /// 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 -- + /// `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 + /// `TraceStream::attach_installation_id_if_present` and + /// `TraceStream::attach_datastore_id_if_present`). + fn refresh_ids(&self) { + self.tracestream + .attach_installation_id_if_present(self.agent_config.datastore_path()); + self.tracestream + .attach_datastore_id_if_present(self.agent_config.datastore_path()); + } + /// Handles a servicing request by acquiring the necessary locks, /// setting up log forwarding, and spawning the provided servicing task. /// @@ -205,6 +231,59 @@ impl TridentServer { // Try to acquire the connection lock in write mode let guard = self.try_acquire_write_lock()?; + // Reject requests that cannot themselves stage a new install/update + // (see `DataStore::may_initialize_datastore_for_command`) when no + // datastore exists yet -- mirrors the CLI's `HostNotProvisioned` + // check in `main.rs`. Without this, e.g. a `commit`/`rollback` RPC + // arriving against an unprovisioned host falls through to + // `DataStore::open_or_create` in the service handler and silently + // creates an empty datastore instead of failing outright. + // Untelemetered, same as the lock-busy rejections above: this is + // admission control, not a distinct servicing outcome. + if !DataStore::may_initialize_datastore_for_command(name) + && !self.agent_config.datastore_path().exists() + { + warn!("Rejected request '{}': datastore does not exist", name); + return Err(Status::failed_precondition("Host is not provisioned")); + } + + // Re-check for a persisted installation ID and datastore ID before + // this request fires its own command_start (below, via + // run_with_operation). server_main's daemon-startup attach only + // ever runs once, at startup -- so a request that arrives before + // any datastore exists (e.g. this daemon's very first install) + // would otherwise never see one, even after that request's own + // handler goes on to create the datastore. + self.refresh_ids(); + + // Tag every metric/tracing event `f` fires (on whatever thread it + // ultimately runs on -- see `spawn_servicing_task`, which runs it + // via `tokio::task::spawn_blocking`, giving it a dedicated OS + // thread for its whole duration) with `command`/`operation_id`, the + // same way the CLI path does for its own dispatch. `name` already + // matches the CLI's own command-naming convention (see + // `command_name` in `main.rs`) for stage/finalize granularity. + // + // If `f` decides a reboot is needed, also capture this operation's + // context (while it's still installed) via `save_reboot_operation`, + // so `server::reboot` -- which runs later, after this whole + // request and even the daemon's main event loop have returned -- + // 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 + }, + ) + }; + // Create the gRPC response channel let (tx, rx) = mpsc::unbounded_channel(); diff --git a/crates/trident/src/validation.rs b/crates/trident/src/validation.rs index ffa59f656..0f5e4242e 100644 --- a/crates/trident/src/validation.rs +++ b/crates/trident/src/validation.rs @@ -9,7 +9,7 @@ use trident_api::{ /// Parse the Host Configuration from a string. Accepts an optional path for /// better error reporting when reading from a file. -pub(crate) fn parse_host_config( +pub fn parse_host_config( contents: &str, path: Option>, ) -> Result { diff --git a/docs/Reference/Agent-Configuration.md b/docs/Reference/Agent-Configuration.md index 92c19c4c4..b9e35f30d 100644 --- a/docs/Reference/Agent-Configuration.md +++ b/docs/Reference/Agent-Configuration.md @@ -1,19 +1,39 @@ ---- -sidebar_position: 5 ---- - -# Agent Configuration - -The Trident Agent Configuration file contains configuration details for Trident. It is used for all Trident commands. The Agent Configuration file path must be `/etc/trident/trident.conf`. - -> In most cases, the default values of Agent Configuration are sufficient and should not need to be changed. - -## Setting Custom Datastore Path - -By default, Trident will use `/var/lib/trident/datastore.sqlite` as the path for the datastore. To configure a non-default path, the Agent Configuration file must contain a line defining the path like this: - -``` conf -DatastorePath=/special/path/to/my-datastore.sqlite -``` - -> The datastore path cannot be hosted on an [A/B volume pair](./Glossary#ab-volume-pair) and must be an absolute path. +--- +sidebar_position: 5 +--- + +# Agent Configuration + +The Trident Agent Configuration file contains configuration details for Trident. It is used for all Trident commands. The Agent Configuration file path must be `/etc/trident/trident.conf`. + +> In most cases, the default values of Agent Configuration are sufficient and should not need to be changed. + +## Setting Custom Datastore Path + +By default, Trident will use `/var/lib/trident/datastore.sqlite` as the path for the datastore. To configure a non-default path, the Agent Configuration file must contain a line defining the path like this: + +``` conf +DatastorePath=/special/path/to/my-datastore.sqlite +``` + +> The datastore path cannot be hosted on an [A/B volume pair](./Glossary#ab-volume-pair) and must be an absolute path. + +## Telemetry + +Trident can optionally send telemetry data to Azure Monitor / Application +Insights. See [Telemetry](./Telemetry.md) for what data is collected and +how delivery failures are handled. This requires an Application Insights +connection string to have been compiled into the Trident binary at build +time (via the `AZURE_MONITOR_CONNECTION_STRING` environment variable); if +no connection string was compiled in, this setting has no effect. + +Telemetry defaults to **disabled** (`OptOut`). To enable it, add a line to +the Agent Configuration file: + +``` conf +Telemetry=OptIn +``` + +The value is case-insensitive (`OptIn`, `optin`, and `OPTIN` are all +equivalent); any value other than a case-insensitive match for `OptIn` +(including an absent `Telemetry` line) is treated as `OptOut`. diff --git a/docs/Reference/Telemetry.md b/docs/Reference/Telemetry.md new file mode 100644 index 000000000..0add1722e --- /dev/null +++ b/docs/Reference/Telemetry.md @@ -0,0 +1,69 @@ +--- +sidebar_position: 4 +--- + +# Telemetry + +Trident records the same metrics/spans locally in two places regardless of +whether remote telemetry is enabled: appended to +`/var/log/trident-metrics.jsonl`, and logged to journald under the +`trident-tracing` syslog identifier. Retrieve the journald copy with: + +``` bash +journalctl -t trident-tracing +``` + +On top of these local copies, Trident can optionally send this same +best-effort stream of tracing data to Azure Monitor / Application +Insights. See [Agent Configuration](./Agent-Configuration.md) for how to +enable it. + +## Host Metadata + +Every event sent also includes as much of the following host metadata as +is available at the time, so operators should be aware this leaves the +host along with the metrics/spans themselves: + +- `asset_id`: the host's DMI product UUID (a stable hardware identifier). +- `os_release`: the `VERSION` field from `/etc/os-release`. +- `kernel_version`: the running kernel release (`uname -r`). +- `total_cpu`: the number of CPUs. +- `total_memory_gib`: total memory, in GiB. +- `trident_version`: the running Trident version. +- `datastore_id`: an ID that lets separate events be correlated back to the + same datastore over its entire lifetime (generated on first access to + the datastore, whether or not an install has actually happened yet). + Not present on events that fire before the datastore has ever been + accessed (e.g. very early in a host's first-ever `install`, before + `Trident::new` opens or creates it). +- `installation_id`: an ID that lets separate events be correlated back to + the same host installation over time. Unlike `datastore_id`, this is + only ever created (get-or-create, never overwritten) at the start of + `Trident::install`. Any event that fires before that point (i.e. + before a host's first-ever install has actually created one) instead + reports that invocation's own `operation_id` as a stand-in + `installation_id` -- the same value that will end up persisted as the + real `installation_id` if that invocation goes on to become the + first-ever install. A datastore created before `installation_id` was + introduced is migrated the first time it is opened: a one-time write + persists an `installation_id` for it (without touching any other + data), so older hosts pick up the field on their next command rather + than remaining permanently without one. +- `operation_id`: an ID that lets events emitted during the same command + invocation be correlated with each other. +- `command`: which command produced the event (e.g. `install`, `update`, + `update_stage`, `update_finalize`, `commit`, `rollback`, `rebuild_raid`). +- `source`: which of Trident's three entry points produced the event -- + `cli` (a command run directly, without a daemon), `daemon` (a command + the daemon executed for a gRPC request), or `grpc-client` (the CLI + acting as a client, relaying a command to a running daemon). + +## Delivery + +Telemetry delivery is always best-effort and never affects servicing +outcomes, but failures are not all logged at the same level: a failure to +serialize an event, or to enqueue it because the background uploader has +already shut down, is logged at trace level, while a failure to actually +deliver an event (e.g. no network connectivity, or a non-2xx response from +Application Insights) is logged at error level, so operators can find +remote-delivery problems in normal logs. diff --git a/packaging/docker/Dockerfile.full b/packaging/docker/Dockerfile.full index d2766bb73..a4da4692b 100644 --- a/packaging/docker/Dockerfile.full +++ b/packaging/docker/Dockerfile.full @@ -36,6 +36,12 @@ ARG TRIDENT_VERSION=dev-build ARG RPM_VER=0.1.0 ARG RPM_REL=1 +# Application Insights connection string identifying telemetry as coming +# from Trident's own CI/CD pipeline builds. Set in pipeline template +# release.yml -- see trident.spec for how this is consumed +# (%{trident_azmon_conn_str}). +ARG AZURE_MONITOR_CONNECTION_STRING="" + ARG RPM_DEST=/usr/src/azl # This entry needs to exist in the config.toml file to allow cargo to use the @@ -52,7 +58,8 @@ RUN --mount=type=secret,id=registry_token \ rpmbuild -bb --build-in-place trident.spec \ --define="trident_version $TRIDENT_VERSION" \ --define="rpm_ver $RPM_VER" \ - --define="rpm_rel $RPM_REL" && \ + --define="rpm_rel $RPM_REL" \ + --define="trident_azmon_conn_str $AZURE_MONITOR_CONNECTION_STRING" && \ tar -czvf trident-rpms.tar.gz -C $RPM_DEST ./RPMS FROM scratch AS artifact diff --git a/packaging/rpm/trident.spec b/packaging/rpm/trident.spec index 9743a1987..de272f2fc 100644 --- a/packaging/rpm/trident.spec +++ b/packaging/rpm/trident.spec @@ -11,6 +11,10 @@ %global selinuxtype targeted +# Azure Monitor / Application Insights connection string compiled into the +# azurelinux distro build of trident binary for best-effort telemetry. +%global trident_azmon_conn_str_public InstrumentationKey=cb38fc09-8473-4b4a-b5e4-208aa66a974f;IngestionEndpoint=https://eastus-8.in.applicationinsights.azure.com/;LiveEndpoint=https://eastus.livediagnostics.monitor.azure.com/;ApplicationId=b9814e3f-a121-4d99-9ba7-eeaf56195c29 + Summary: Declarative, security-first OS lifecycle agent designed primarily for Azure Linux Name: trident # Use hard-coded versions for distro build @@ -273,9 +277,25 @@ EOF %if %{undefined rpm_ver} # Use %{version}-%{release} for TRIDENT_VERSION in distro build export TRIDENT_VERSION="%{version}-%{release}" +# Public-usage placeholder connection string (see comment near the top of +# this spec file). +export AZURE_MONITOR_CONNECTION_STRING="%{trident_azmon_conn_str_public}" %else # Use %{trident_version} for Trident repo build export TRIDENT_VERSION="%{trident_version}" +# Connection string identifying telemetry as coming from Trident's own +# CI/CD pipeline builds (as opposed to azurelinux distro-package installs, +# which use the different, hardcoded connection string above). Hardcoded in +# .pipelines/templates/stages/trident_rpms/release.yml and passed through +# to this spec as an rpmbuild --define, the same way %{trident_version} is. +# +# Use the optional-expansion form (`%{?...}`): repo-build paths that define +# rpm_ver but do not pass --define trident_azmon_conn_str (e.g. +# packaging/docker/Dockerfile.full.public) must fall back to an empty +# string, matching the documented no-telemetry default -- not the literal, +# undefined `%{trident_azmon_conn_str}` text RPM would otherwise leave in +# place. +export AZURE_MONITOR_CONNECTION_STRING="%{?trident_azmon_conn_str}" %endif cargo build --release -p trident -p trident-acl-agent