telemetry [11]: report daemon dispatch and request errors

crates/trident/src/logging/operation_context.rs

@@ -43,6 +43,19 @@ use trident_api::{config::Operations, error::TridentError};
 
 use crate::command_kind::CommandKind;
 
+/// Message used by `server::tridentserver::TridentServer::try_acquire_read_lock`/
+/// `try_acquire_write_lock` for the `Status::unavailable` returned when
+/// connection-lock contention blocks a request. `grpc_client::is_transport_failure`
+/// matches on this exact message to recognize this as a deliberate
+/// admission-control rejection -- the daemon DID receive and answer the
+/// request -- rather than a genuine transport-level failure, even though
+/// tonic uses the same `Code::Unavailable` for both.
+pub(crate) const CONNECTION_LOCK_BUSY_MESSAGE: &str = "Trident is busy";
+
+/// Same as [`CONNECTION_LOCK_BUSY_MESSAGE`], but for the servicing-lock
+/// contention rejections in `servicing_request`/`reading_request`.
+pub(crate) const SERVICING_LOCK_BUSY_MESSAGE: &str = "Servicing is active";
+
 /// Maps a base command name plus its requested `Operations` to the same
 /// naming convention gRPC's `servicing_request` already uses for
 /// stage/finalize granularity (e.g. `"install"` vs `"install_stage"` vs

crates/trident/src/logging/tracestream.rs

@@ -241,14 +241,26 @@ impl TraceStream {
     /// servicing ID yet), so it's still worth attaching even when
     /// `skip_installation_id` is true.
     ///
-    /// Safe to call from anywhere, any number of times, before any point
-    /// that wants either ID attached: this is the single implementation
-    /// shared by every read-only-in-the-common-case attach call site (the
-    /// CLI's dispatch, the daemon's startup attach, and the daemon's
-    /// per-request backstop), so a correctness fix to this logic only
-    /// needs to happen once.
+    /// Refresh servicing IDs only before servicing starts or while holding
+    /// its lock. Rejected requests must use the installation-only path so
+    /// a stale datastore read cannot overwrite an active operation's ID.
     /// Known edge case: multiboot finalize may prewarm the source `servicing_id`.
     pub fn attach_ids_if_present(&self, datastore_path: &Path, skip_installation_id: bool) {
+        self.attach_ids_from_path(datastore_path, skip_installation_id, true);
+    }
+
+    /// Attaches or migrates the installation ID without touching the shared
+    /// servicing ID. Safe for rejected requests outside the servicing lock.
+    pub fn attach_installation_id_if_present(&self, datastore_path: &Path) {
+        self.attach_ids_from_path(datastore_path, false, false);
+    }
+
+    fn attach_ids_from_path(
+        &self,
+        datastore_path: &Path,
+        skip_installation_id: bool,
+        attach_servicing_id: bool,
+    ) {
         if !datastore_path.exists() {
             return;
         }
@@ -275,7 +287,9 @@ impl TraceStream {
             }
         }
 
-        self.attach_servicing_id_if_present(&datastore);
+        if attach_servicing_id {
+            self.attach_servicing_id_if_present(&datastore);
+        }
     }
 
     /// Ensures `datastore` has an installation ID -- creating one if this
@@ -987,6 +1001,34 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_installation_only_attachment_preserves_active_servicing_id() {
+        let directory = tempfile::tempdir().unwrap();
+        let path = directory.path().join("datastore.sqlite");
+        let mut datastore = DataStore::open_or_create(&path).unwrap();
+        let installation = datastore.ensure_installation_id().unwrap().to_string();
+        let persisted_servicing = datastore.ensure_servicing_id().unwrap().to_string();
+        drop(datastore);
+
+        let stream = TraceStream::default();
+        stream.set_servicing_id("active-stage".to_string());
+        stream.attach_installation_id_if_present(&path);
+        assert_eq!(
+            *stream.installation_id_handle().read().unwrap(),
+            Some(installation)
+        );
+        assert_eq!(
+            *stream.servicing_id_handle().read().unwrap(),
+            Some("active-stage".to_string())
+        );
+
+        stream.attach_ids_if_present(&path, false);
+        assert_eq!(
+            *stream.servicing_id_handle().read().unwrap(),
+            Some(persisted_servicing)
+        );
+    }
+
     #[test]
     fn test_tracestream() {
         let temp_dir = tempfile::tempdir().unwrap();

crates/trident/src/server/tridentserver/mod.rs

@@ -160,21 +160,33 @@ impl TridentServer {
     /// Tries to acquire a read lock on the server's RwLock. If the lock
     /// cannot be acquired, returns a gRPC Status indicating that the server is
     /// busy.
+    ///
+    /// Intentionally fires no telemetry (`command_start`/`command_error`) for
+    /// this rejection, unlike `reject_invalid_argument`/`reject_invalid_field`:
+    /// a connection-lock contention failure isn't a
+    /// distinct servicing outcome the way a malformed request is -- it's
+    /// pure admission control, happens before `refresh_ids` would
+    /// even run, and (unlike a bad request) the caller is expected to retry
+    /// the exact same request rather than fix anything, so a low-value,
+    /// high-volume `command_error` stream isn't worth adding here.
     #[cfg(feature = "grpc-preview")]
     fn try_acquire_read_lock(&self) -> Result<OwnedRwLockReadGuard<()>, Status> {
         self.rwlock.clone().try_read_owned().map_err(|_| {
             warn!("Trident is busy, cannot acquire read connection lock");
-            Status::unavailable("Trident is busy")
+            Status::unavailable(operation_context::CONNECTION_LOCK_BUSY_MESSAGE)
         })
     }
 
     /// Tries to acquire a write lock on the server's RwLock. If the lock
     /// cannot be acquired, returns a gRPC Status indicating that the server is
     /// busy.
+    ///
+    /// See the telemetry note on [`Self::try_acquire_read_lock`]: this
+    /// rejection is intentionally untelemetered for the same reason.
     fn try_acquire_write_lock(&self) -> Result<OwnedRwLockWriteGuard<()>, Status> {
         self.rwlock.clone().try_write_owned().map_err(|_| {
             warn!("Trident is busy, cannot acquire write connection lock");
-            Status::unavailable("Trident is busy")
+            Status::unavailable(operation_context::CONNECTION_LOCK_BUSY_MESSAGE)
         })
     }
 
@@ -259,14 +271,14 @@ impl TridentServer {
 
         // Re-check for a persisted installation ID and current servicing
         // ID before this request fires its own command_start (below, via
-        // run_with_operation). server_main's daemon-startup attach only
+        // run_command). server_main's daemon-startup attach only
         // ever runs once, at startup -- so a request that arrives before
         // any datastore exists (e.g. this daemon's very first install)
         // would otherwise never see one, even after that request's own
         // handler goes on to create the datastore.
         //
         // Deferred into the `f` closure below (called right before
-        // run_with_operation, preserving "before command_start") rather
+        // run_command, preserving "before command_start") rather
         // than run synchronously here: `attach_ids_if_present` opens the
         // datastore, and datastore connections use a 5-second busy
         // timeout, so lock contention could otherwise stall this
@@ -297,29 +309,33 @@ impl TridentServer {
         let kind_for_closure = kind.clone();
         let f = move || {
             Self::refresh_ids(&tracestream_for_ids, &datastore_path_for_ids);
-            operation_context::run_with_operation(
-                &kind_for_closure,
-                operation_context::OperationSource::Daemon,
-                || {
-                    let result = f();
-                    if let Ok((ExitKind::NeedsReboot, ..)) = &result {
-                        operation_context::save_reboot_operation();
-                    }
-                    result
-                },
-            )
+            run_servicing_command(&kind_for_closure, || {
+                let result = f();
+                if let Ok((ExitKind::NeedsReboot, ..)) = &result {
+                    operation_context::save_reboot_operation();
+                }
+                result
+            })
         };
 
         // Create the gRPC response channel
         let (tx, rx) = mpsc::unbounded_channel();
 
-        // Try to acquire the servicing lock
+        // Try to acquire the servicing lock. Rejected here, after
+        // `refresh_ids` above but before the `run_command`
+        // closure `f` (built above) ever runs, this is intentionally
+        // untelemetered for the same reason as the connection-lock
+        // rejections in `try_acquire_read_lock`/`try_acquire_write_lock`:
+        // it's admission control, not a distinct servicing outcome, and
+        // the caller is expected to retry rather than fix anything.
         let Some(servicing_guard) = self.servicing_manager.try_lock_servicing() else {
             warn!(
                 "Request '{}' blocked because servicing is active",
                 kind.as_str()
             );
-            return Err(Status::unavailable("Servicing is active"));
+            return Err(Status::unavailable(
+                operation_context::SERVICING_LOCK_BUSY_MESSAGE,
+            ));
         };
 
         // Set up log forwarding. Logs will be sent over the gRPC channel.
@@ -423,15 +439,31 @@ impl TridentServer {
         // request.
         let _guard = self.try_acquire_read_lock()?;
 
-        // Try to acquire the servicing read lock
+        // Try to acquire the servicing read lock. Same intentional
+        // telemetry gap as the connection-lock rejections in
+        // `try_acquire_read_lock`/`try_acquire_write_lock` and the
+        // servicing-lock rejection in `servicing_request`: it's admission
+        // control rather than a distinct read outcome, and the caller is
+        // expected to retry rather than fix anything.
         let Some(servicing_guard) = self.servicing_manager.try_lock_reading() else {
             warn!(
                 "Read request '{}' blocked because servicing is active",
                 name
             );
-            return Err(Status::unavailable("Servicing is active"));
+            return Err(Status::unavailable(
+                operation_context::SERVICING_LOCK_BUSY_MESSAGE,
+            ));
         };
 
+        // Read requests (e.g. `get_servicing_state`, `check_rollback`) are
+        // intentionally left untelemetered -- like their CLI counterparts
+        // (`get`, `validate`, `diagnose`, etc.; see `run_trident` in
+        // main.rs), none of them emit `command_start`/`command_error` or
+        // any other `metric_name` event, so there's no need to prep the
+        // TraceStream's installation ID/database ID (`refresh_ids`, used
+        // by `servicing_request` for exactly that reason) or wrap `f` in
+        // `operation_context::run_command` -- it just runs directly here.
+
         // Execute the reading function
         Ok(Response::new(
             ServicingManager::spawn_reading_task(servicing_guard, f)
@@ -444,6 +476,15 @@ impl TridentServer {
     }
 }
 
+fn run_servicing_command<T>(
+    kind: &CommandKind,
+    f: impl FnOnce() -> Result<T, TridentError>,
+) -> Result<T, TridentError> {
+    operation_context::run_command_if(kind, operation_context::OperationSource::Daemon, f, |_| {
+        kind.is_servicing()
+    })
+}
+
 #[cfg(feature = "grpc-preview")]
 fn trident_error_to_status(err: TridentError) -> Status {
     let code = match err.kind() {
@@ -458,3 +499,72 @@ fn trident_error_to_status(err: TridentError) -> Status {
 
     Status::new(code, format!("{err:?}"))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use tracing::{field::Visit, Event, Subscriber};
+    use tracing_subscriber::{layer::Context, prelude::*, Layer, Registry};
+
+    #[derive(Clone, Default)]
+    struct CommandErrors(Arc<AtomicUsize>);
+
+    impl<S: Subscriber> Layer<S> for CommandErrors {
+        fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
+            struct Metric(bool);
+            impl Visit for Metric {
+                fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
+                    if field.name() == "metric_name" && value == "command_error" {
+                        self.0 = true;
+                    }
+                }
+
+                fn record_debug(&mut self, _: &tracing::field::Field, _: &dyn std::fmt::Debug) {}
+            }
+            let mut metric = Metric(false);
+            event.record(&mut metric);
+            if metric.0 {
+                self.0.fetch_add(1, Ordering::SeqCst);
+            }
+        }
+    }
+
+    #[test]
+    fn test_request_errors_follow_servicing_classification() {
+        for kind in [CommandKind::check_root(), CommandKind::commit()] {
+            let errors = CommandErrors::default();
+            let subscriber = Registry::default().with(errors.clone());
+            let result: Result<(), TridentError> =
+                tracing::subscriber::with_default(subscriber, || {
+                    run_servicing_command(&kind, || {
+                        let (_, active, source) = operation_context::current().unwrap();
+                        assert_eq!(active.as_str(), kind.as_str());
+                        assert_eq!(source, operation_context::OperationSource::Daemon);
+                        Err(TridentError::internal("request failed"))
+                    })
+                });
+            assert!(result.is_err());
+            assert_eq!(
+                errors.0.load(Ordering::SeqCst),
+                usize::from(kind.is_servicing())
+            );
+            assert!(operation_context::current().is_none());
+        }
+    }
+
+    #[test]
+    fn test_nonservicing_request_panic_is_still_reported() {
+        let errors = CommandErrors::default();
+        let subscriber = Registry::default().with(errors.clone());
+        let result = tracing::subscriber::with_default(subscriber, || {
+            panic::catch_unwind(|| {
+                let _: Result<(), TridentError> =
+                    run_servicing_command(&CommandKind::check_root(), || panic!("request panic"));
+            })
+        });
+        assert!(result.is_err());
+        assert_eq!(errors.0.load(Ordering::SeqCst), 1);
+        assert!(operation_context::current().is_none());
+    }
+}

crates/trident/src/server/tridentserver/services/install.rs

@@ -27,17 +27,27 @@ impl InstallService for TridentServer {
     ) -> Result<Response<Self::InstallStream>, Status> {
         let req = request.into_inner();
         let Some(staging) = req.stage else {
-            return Err(Status::invalid_argument("Missing staging configuration"));
+            return Err(self.reject_invalid_argument(
+                &CommandKind::install(),
+                "stage",
+                "Missing staging configuration",
+            ));
         };
 
         let Some(host_config) = staging.config else {
-            return Err(Status::invalid_argument(
+            return Err(self.reject_invalid_argument(
+                &CommandKind::install(),
+                "stage.config",
                 "Missing host configuration in staging configuration",
             ));
         };
 
         let Some(finalize) = req.finalize else {
-            return Err(Status::invalid_argument("Missing finalize configuration"));
+            return Err(self.reject_invalid_argument(
+                &CommandKind::install(),
+                "finalize",
+                "Missing finalize configuration",
+            ));
         };
 
         let data_store_path = self.agent_config.datastore_path().to_owned();
@@ -74,7 +84,9 @@ impl InstallService for TridentServer {
         let req = request.into_inner();
 
         let Some(host_config) = req.config else {
-            return Err(Status::invalid_argument(
+            return Err(self.reject_invalid_argument(
+                &CommandKind::install_stage(),
+                "config",
                 "Missing host configuration in staging configuration",
             ));
         };