feat(service): Add span instrumentation to backends (#542)

Backend operations now appear as real child spans in Sentry, turning the
existing per-request and per-task transactions into full trace
waterfalls. They were empty before because the sentry-tracing layer only
converts `INFO`+ spans by default; the logging layer now forwards
`DEBUG`+ spans (`TRACE` still ignored) and the backend spans moved to
`debug`.

GCS/S3 HTTP calls and BigTable RPCs each get a per-attempt span covering
the real request duration (method/URL/status, or action/gRPC status),
and the tiered, changelog, and per-backend operations are annotated so
the waterfall reads top-down. Spans carry small primitive fields (object
id, range, part numbers, URLs); payloads, streams, and metadata are
excluded.

Observability-only — no behavior changes and console logging is
untouched.

Author: jan-auer

objectstore-log/src/subscriber.rs

@@ -15,21 +15,24 @@ include!(concat!(env!("OUT_DIR"), "/constants.gen.rs"));
 ///
 /// When built with the `sentry` feature, also attaches a Sentry tracing layer if the Sentry client
 /// has been initialized. `ERROR` and `WARN` events become Sentry events; `INFO`/`DEBUG` become
-/// Sentry logs; `TRACE` is ignored.
+/// Sentry logs; `TRACE` events are ignored. `ERROR`..`DEBUG` spans become Sentry spans; `TRACE`
+/// spans are ignored.
 pub fn init(config: &LoggingConfig) {
     #[cfg(feature = "sentry")]
     let sentry_layer = sentry::Hub::current()
         .client()
         .filter(|c| c.is_enabled())
         .map(|_| {
             use sentry::integrations::tracing as sentry_tracing;
-            sentry_tracing::layer().event_filter(|metadata| match *metadata.level() {
-                Level::ERROR | Level::WARN => {
-                    sentry_tracing::EventFilter::Event | sentry_tracing::EventFilter::Log
-                }
-                Level::INFO | Level::DEBUG => sentry_tracing::EventFilter::Log,
-                Level::TRACE => sentry_tracing::EventFilter::Ignore,
-            })
+            sentry_tracing::layer()
+                .event_filter(|metadata| match *metadata.level() {
+                    Level::ERROR | Level::WARN => {
+                        sentry_tracing::EventFilter::Event | sentry_tracing::EventFilter::Log
+                    }
+                    Level::INFO | Level::DEBUG => sentry_tracing::EventFilter::Log,
+                    Level::TRACE => sentry_tracing::EventFilter::Ignore,
+                })
+                .span_filter(|metadata| !matches!(*metadata.level(), Level::TRACE))
         });
 
     let format = tracing_subscriber::fmt::layer()

objectstore-service/src/backend/bigtable.rs

@@ -41,6 +41,7 @@ use objectstore_types::metadata::{ExpirationPolicy, Metadata};
 use objectstore_types::range::{ByteRange, ContentRange};
 use serde::{Deserialize, Serialize};
 use tonic::Code;
+use tracing::Instrument;
 
 use crate::backend::common::{
     Backend, DeleteResponse, GetResponse, HighVolumeBackend, MetadataResponse, PutResponse,
@@ -740,6 +741,7 @@ impl BigTableBackend {
     /// Reads a single row by key, returning parsed row data.
     ///
     /// Returns `None` if the row is absent or has expired.
+    #[tracing::instrument(level = "debug", fields(action), skip_all)]
     async fn read_row(
         &self,
         path: &[u8],
@@ -776,6 +778,7 @@ impl BigTableBackend {
         })
     }
 
+    #[tracing::instrument(level = "debug", fields(action), skip_all)]
     async fn mutate(
         &self,
         path: Vec<u8>,
@@ -821,6 +824,7 @@ impl BigTableBackend {
     /// Best-effort TTI bump for a row.
     ///
     /// If the payload isn't loaded, it will be fetched. Failures are ignored silently.
+    #[tracing::instrument(level = "debug", fields(?hv_id, loaded), skip_all)]
     async fn bump_tti(&self, path: Vec<u8>, row: &RowData, loaded: bool, hv_id: &ObjectId) {
         let expiration_policy = row.expiration_policy();
 
@@ -860,6 +864,7 @@ impl BigTableBackend {
     }
 
     /// Executes a `CheckAndMutateRow` request.
+    #[tracing::instrument(level = "debug", fields(action = context), skip_all)]
     async fn check_and_mutate(
         &self,
         row_key: Vec<u8>,
@@ -898,7 +903,7 @@ impl Backend for BigTableBackend {
         "bigtable"
     }
 
-    #[tracing::instrument(level = "trace", fields(?id), skip_all)]
+    #[tracing::instrument(level = "debug", fields(?id), skip_all)]
     async fn put_object(
         &self,
         id: &ObjectId,
@@ -919,7 +924,7 @@ impl Backend for BigTableBackend {
         Ok(())
     }
 
-    #[tracing::instrument(level = "trace", fields(?id), skip_all)]
+    #[tracing::instrument(level = "debug", skip(self))]
     async fn get_object(&self, id: &ObjectId, range: Option<ByteRange>) -> Result<GetResponse> {
         match self.get_tiered_object(id, range).await? {
             TieredGet::Object(metadata, content_range, payload) => {
@@ -930,7 +935,7 @@ impl Backend for BigTableBackend {
         }
     }
 
-    #[tracing::instrument(level = "trace", fields(?id), skip_all)]
+    #[tracing::instrument(level = "debug", skip(self))]
     async fn get_metadata(&self, id: &ObjectId) -> Result<MetadataResponse> {
         match self.get_tiered_metadata(id).await? {
             TieredMetadata::Object(metadata) => Ok(Some(metadata)),
@@ -939,7 +944,7 @@ impl Backend for BigTableBackend {
         }
     }
 
-    #[tracing::instrument(level = "trace", fields(?id), skip_all)]
+    #[tracing::instrument(level = "debug", skip(self))]
     async fn delete_object(&self, id: &ObjectId) -> Result<DeleteResponse> {
         objectstore_log::debug!("Deleting from Bigtable backend");
 
@@ -952,7 +957,7 @@ impl Backend for BigTableBackend {
 
 #[async_trait::async_trait]
 impl HighVolumeBackend for BigTableBackend {
-    #[tracing::instrument(level = "trace", fields(?id), skip_all)]
+    #[tracing::instrument(level = "debug", fields(?id), skip_all)]
     async fn put_non_tombstone(
         &self,
         id: &ObjectId,
@@ -1000,7 +1005,7 @@ impl HighVolumeBackend for BigTableBackend {
         Err(Error::generic("BigTable: race loop in put_non_tombstone"))
     }
 
-    #[tracing::instrument(level = "trace", fields(?id), skip_all)]
+    #[tracing::instrument(level = "debug", skip(self))]
     async fn get_tiered_object(
         &self,
         id: &ObjectId,
@@ -1037,7 +1042,7 @@ impl HighVolumeBackend for BigTableBackend {
         })
     }
 
-    #[tracing::instrument(level = "trace", fields(?id), skip_all)]
+    #[tracing::instrument(level = "debug", skip(self))]
     async fn get_tiered_metadata(&self, id: &ObjectId) -> Result<TieredMetadata> {
         objectstore_log::debug!("Reading metadata from Bigtable backend");
         let path = id.as_storage_path().to_string().into_bytes();
@@ -1066,7 +1071,7 @@ impl HighVolumeBackend for BigTableBackend {
         })
     }
 
-    #[tracing::instrument(level = "trace", fields(?id), skip_all)]
+    #[tracing::instrument(level = "debug", skip(self))]
     async fn delete_non_tombstone(&self, id: &ObjectId) -> Result<Option<Tombstone>> {
         objectstore_log::debug!("Conditional delete from Bigtable backend");
 
@@ -1110,7 +1115,7 @@ impl HighVolumeBackend for BigTableBackend {
         ))
     }
 
-    #[tracing::instrument(level = "trace", fields(?id), skip_all)]
+    #[tracing::instrument(level = "debug", skip(self, write))]
     async fn compare_and_write(
         &self,
         id: &ObjectId,
@@ -1205,7 +1210,26 @@ where
     let mut retry_count = 0usize;
 
     loop {
-        match f().await {
+        let attempt_span = tracing::debug_span!(
+            "bigtable.request",
+            action = context,
+            grpc.status = tracing::field::Empty,
+        );
+        let attempt = async {
+            let result = f().await;
+            let span = tracing::Span::current();
+            match &result {
+                Ok(_) => span.record("grpc.status", "ok"),
+                Err(BigTableError::RpcError(status)) => {
+                    span.record("grpc.status", tracing::field::debug(status.code()))
+                }
+                // Non-RPC error; the error event carries the details.
+                Err(_) => &span,
+            };
+            result
+        };
+
+        match attempt.instrument(attempt_span).await {
             Ok(res) => return Ok(res),
             Err(e) if retry_count >= REQUEST_RETRY_COUNT || !is_retryable(&e) => {
                 objectstore_metrics::count!("bigtable.failures", action = context);

objectstore-service/src/backend/changelog.rs

@@ -121,6 +121,7 @@ impl ChangeManager {
     ///
     /// When the [`ChangeGuard`] is dropped, a background process is spawned to
     /// clean up any unreferenced objects in LT storage.
+    #[tracing::instrument(level = "debug", fields(id = ?change.id, new = ?change.new, old = ?change.old), skip_all)]
     pub async fn record(self: Arc<Self>, change: Change) -> Result<ChangeGuard> {
         let token = self.tracker.token();
 
@@ -143,6 +144,7 @@ impl ChangeManager {
     /// Behaves like [`Self::record`], except that the guard is created in the `Assembling` state.
     /// Unlike other states, this guard does nothing on drop, leaving the burden of cleaning up to
     /// the [`ChangeLog`].
+    #[tracing::instrument(level = "debug", fields(id = ?change.id, new = ?change.new, old = ?change.old), skip_all)]
     pub async fn record_assembling(self: Arc<Self>, change: Change) -> Result<ChangeGuard> {
         let token = self.tracker.token();
 

…ectstore-service/src/backend/response.rs …tstore-service/src/backend/extensions.rsobjectstore-service/src/backend/response.rs renamed to objectstore-service/src/backend/extensions.rs objectstore-service/src/backend/response.rs renamed to objectstore-service/src/backend/extensions.rs

@@ -1,16 +1,46 @@
-//! HTTP response status checking with error body parsing.
+//! Extension traits for `reqwest` requests and responses.
 //!
-//! Provides [`ResponseExt`], an extension trait that replaces
-//! [`reqwest::Response::error_for_status`] with a version that reads the
-//! response body on 4xx/5xx errors and parses the structured error code and
-//! message from it (JSON for GCS JSON API, XML for GCS XML API and S3).
+//! Provides [`SendTraced`], which sends a request inside a tracing span, and
+//! [`ResponseExt`], which replaces [`reqwest::Response::error_for_status`] with a
+//! version that reads the response body on 4xx/5xx errors and parses the
+//! structured error code and message from it (JSON for GCS JSON API, XML for GCS
+//! XML API and S3).
 
 use reqwest::{Response, header};
 use serde::Deserialize;
+use tracing::Instrument;
 
 use crate::error::{BackendDetail, Error, Result};
 use crate::stream;
 
+/// Extension trait that sends a request inside a tracing span.
+pub trait SendTraced {
+    /// Sends the request, wrapping it in a span that covers the full request
+    /// duration and records the response status code.
+    async fn send_traced(self) -> reqwest::Result<reqwest::Response>;
+}
+
+impl SendTraced for reqwest::RequestBuilder {
+    async fn send_traced(self) -> reqwest::Result<reqwest::Response> {
+        let (client, request) = self.build_split();
+        let request = request?;
+        let span = tracing::debug_span!(
+            "http.request",
+            method = %request.method(),
+            url = %request.url(),
+            http.status_code = tracing::field::Empty,
+        );
+        let send_future = async {
+            let response = client.execute(request).await;
+            if let Ok(response) = &response {
+                tracing::Span::current().record("http.status_code", response.status().as_u16());
+            }
+            response
+        };
+        send_future.instrument(span).await
+    }
+}
+
 /// GCS JSON API error envelope (`{"error": {"message": "...", ...}}`).
 #[derive(Deserialize)]
 struct JsonApiError {