feat: surface ObjectStoreOperation on outbound HTTP requests (HTTP backend)

src/client/builder.rs

@@ -98,6 +98,22 @@ impl HttpRequestBuilder {
         self
     }
 
+    /// Attach an [`ObjectStoreOperation`](crate::client::ObjectStoreOperation)
+    /// to the outbound request via `http::Request::extensions`. Wrapping
+    /// `HttpService` impls (tower layers, reqwest middleware, etc.) can
+    /// read it via `req.extensions().get::<ObjectStoreOperation>()` to
+    /// produce useful trace spans without sniffing URLs.
+    ///
+    /// Marked `allow(dead_code)` because only the `http` backend wires
+    /// it up in this PR; AWS / GCP / Azure follow-up PRs will use it.
+    #[allow(dead_code)]
+    pub(crate) fn operation(mut self, op: crate::client::ObjectStoreOperation) -> Self {
+        if let Ok(r) = &mut self.request {
+            r.extensions_mut().insert(op);
+        }
+        self
+    }
+
     pub(crate) fn header<K, V>(mut self, name: K, value: V) -> Self
     where
         K: TryInto<HeaderName>,

src/client/http/mod.rs

@@ -25,3 +25,6 @@ pub use connection::*;
 
 mod spawn;
 pub use spawn::*;
+
+mod operation;
+pub use operation::{ObjectStoreOperation, OperationKind};

src/client/http/operation.rs

@@ -0,0 +1,108 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Per-call operation context surfaced via `http::Request::extensions`.
+//!
+//! Each backend inserts an [`ObjectStoreOperation`] into the outbound
+//! [`HttpRequest`]'s extensions before it reaches an [`HttpService`]. A
+//! wrapping `HttpService` (or `tower::Service` / `reqwest_middleware::Middleware`)
+//! can then read it via `req.extensions().get::<ObjectStoreOperation>()` to
+//! produce meaningful trace spans without sniffing URLs and headers.
+//!
+//! [`http::Extensions`] are in-process only and never reach the wire, so this
+//! is invisible to remote servers.
+//!
+//! [`HttpRequest`]: crate::client::HttpRequest
+//! [`HttpService`]: crate::client::HttpService
+
+use crate::path::Path;
+
+/// Identifies the high-level `ObjectStore` operation that produced an
+/// outbound HTTP request.
+///
+/// New variants may be added without a major version bump (`#[non_exhaustive]`).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+#[non_exhaustive]
+pub enum OperationKind {
+    /// `ObjectStore::get` / `get_opts` / `get_range` / `get_ranges`
+    Get,
+    /// `ObjectStore::head`
+    Head,
+    /// `ObjectStore::put` / `put_opts`
+    Put,
+    /// Any phase of a `put_multipart` upload (create, upload-part, complete, abort)
+    PutMultipart,
+    /// `ObjectStore::delete`
+    Delete,
+    /// `ObjectStore::copy` / `copy_if_not_exists` / `rename` / `rename_if_not_exists`
+    Copy,
+    /// `ObjectStore::list` / `list_with_offset`
+    List,
+    /// `ObjectStore::list_with_delimiter`
+    ListWithDelimiter,
+}
+
+impl OperationKind {
+    /// A short, snake_case identifier suitable for span attributes.
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::Get => "get",
+            Self::Head => "head",
+            Self::Put => "put",
+            Self::PutMultipart => "put_multipart",
+            Self::Delete => "delete",
+            Self::Copy => "copy",
+            Self::List => "list",
+            Self::ListWithDelimiter => "list_with_delimiter",
+        }
+    }
+}
+
+/// Per-call operation context attached to outbound HTTP requests via
+/// [`http::Extensions`].
+///
+/// New fields may be added without a major version bump (`#[non_exhaustive]`).
+#[derive(Debug, Clone)]
+#[non_exhaustive]
+pub struct ObjectStoreOperation {
+    /// The high-level operation that triggered this request.
+    pub kind: OperationKind,
+    /// The user-supplied [`Path`], when applicable. `None` for operations
+    /// that don't take a path argument (e.g. bulk delete, list root, copy
+    /// per-side, multipart sub-requests where the path is implicit).
+    pub location: Option<Path>,
+    /// The backend that issued the request: `"s3"`, `"gcs"`, `"azure"`,
+    /// or `"http"`.
+    pub backend: &'static str,
+}
+
+impl ObjectStoreOperation {
+    /// Construct a new [`ObjectStoreOperation`].
+    pub fn new(kind: OperationKind, backend: &'static str) -> Self {
+        Self {
+            kind,
+            location: None,
+            backend,
+        }
+    }
+
+    /// Attach a [`Path`] to this operation.
+    pub fn with_location(mut self, location: Path) -> Self {
+        self.location = Some(location);
+        self
+    }
+}

src/http/client.rs

@@ -19,7 +19,9 @@ use super::STORE;
 use crate::client::get::GetClient;
 use crate::client::header::HeaderConfig;
 use crate::client::retry::{self, RetryConfig, RetryContext, RetryExt};
-use crate::client::{GetOptionsExt, HttpClient, HttpError, HttpResponse};
+use crate::client::{
+    GetOptionsExt, HttpClient, HttpError, HttpResponse, ObjectStoreOperation, OperationKind,
+};
 use crate::path::{DELIMITER, Path};
 use crate::util::deserialize_rfc1123;
 use crate::{Attribute, Attributes, ClientOptions, GetOptions, ObjectMeta, PutPayload, Result};
@@ -181,7 +183,10 @@ impl Client {
         let mut retry = false;
         loop {
             let url = self.path_url(location);
-            let mut builder = self.client.put(url);
+            let mut builder = self.client.put(url).operation(
+                ObjectStoreOperation::new(OperationKind::Put, "http")
+                    .with_location(location.clone()),
+            );
 
             let mut has_content_type = false;
             for (k, v) in &attributes {
@@ -247,9 +252,14 @@ impl Client {
             .unwrap_or_else(|| self.url.to_string());
 
         let method = Method::from_bytes(b"PROPFIND").unwrap();
+        let mut op = ObjectStoreOperation::new(OperationKind::List, "http");
+        if let Some(path) = location {
+            op = op.with_location(path.clone());
+        }
         let result = self
             .client
             .request(method, url)
+            .operation(op)
             .header("Depth", depth)
             .retryable(&self.retry_config)
             .idempotent(true)
@@ -296,6 +306,10 @@ impl Client {
         let url = self.path_url(path);
         self.client
             .delete(url)
+            .operation(
+                ObjectStoreOperation::new(OperationKind::Delete, "http")
+                    .with_location(path.clone()),
+            )
             .send_retry(&self.retry_config)
             .await
             .map_err(|source| source.error(STORE, path.to_string()))?;
@@ -310,6 +324,10 @@ impl Client {
             let mut builder = self
                 .client
                 .request(method, self.path_url(from))
+                .operation(
+                    ObjectStoreOperation::new(OperationKind::Copy, "http")
+                        .with_location(from.clone()),
+                )
                 .header("Destination", self.path_url(to).as_str());
 
             if !overwrite {
@@ -370,15 +388,19 @@ impl GetClient for Client {
         options: GetOptions,
     ) -> Result<HttpResponse> {
         let url = self.path_url(path);
-        let method = match options.head {
-            true => Method::HEAD,
-            false => Method::GET,
+        let (method, kind) = match options.head {
+            true => (Method::HEAD, OperationKind::Head),
+            false => (Method::GET, OperationKind::Get),
         };
         let has_range = options.range.is_some();
-        let builder = self.client.request(method, url);
-
-        let res = builder
+        // `.operation(...)` must come AFTER `.with_get_options(...)` because the
+        // latter calls `.extensions(...)` which overwrites the entire
+        // `http::Extensions`, wiping anything set earlier in the chain.
+        let res = self
+            .client
+            .request(method, url)
             .with_get_options(options)
+            .operation(ObjectStoreOperation::new(kind, "http").with_location(path.clone()))
             .retryable_request()
             .send(ctx)
             .await
@@ -516,3 +538,104 @@ pub(crate) struct Prop {
 pub(crate) struct ResourceType {
     collection: Option<()>,
 }
+
+// `mock_server` is only built `#[cfg(not(target_arch = "wasm32"))]`, so the
+// test module that uses it must follow the same gate.
+#[cfg(all(test, not(target_arch = "wasm32")))]
+mod tests {
+    use super::*;
+    use crate::client::get::GetClient;
+    use crate::client::mock_server::MockServer;
+    use crate::client::retry::RetryContext;
+    use crate::client::{HttpClient, HttpRequest, HttpResponse, HttpService};
+    use async_trait::async_trait;
+    use hyper::Response;
+    use parking_lot::Mutex;
+    use std::sync::Arc;
+
+    /// `HttpService` wrapper that captures any [`ObjectStoreOperation`]
+    /// found on outbound requests, then forwards to an inner reqwest client.
+    #[derive(Debug)]
+    struct RecordingService {
+        inner: reqwest::Client,
+        seen: Arc<Mutex<Vec<ObjectStoreOperation>>>,
+    }
+
+    #[async_trait]
+    impl HttpService for RecordingService {
+        async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError> {
+            if let Some(op) = req.extensions().get::<ObjectStoreOperation>() {
+                self.seen.lock().push(op.clone());
+            }
+            self.inner.call(req).await
+        }
+    }
+
+    async fn setup() -> (Client, Arc<Mutex<Vec<ObjectStoreOperation>>>, MockServer) {
+        let mock = MockServer::new().await;
+        let seen = Arc::new(Mutex::new(Vec::new()));
+        let recording = RecordingService {
+            inner: reqwest::Client::new(),
+            seen: Arc::clone(&seen),
+        };
+        let http_client = HttpClient::new(recording);
+        let client = Client::new(
+            Url::parse(mock.url()).unwrap(),
+            http_client,
+            ClientOptions::default(),
+            RetryConfig::default(),
+        );
+        (client, seen, mock)
+    }
+
+    #[tokio::test]
+    async fn delete_populates_operation_extension() {
+        let (client, seen, mock) = setup().await;
+        mock.push(Response::new("".to_string()));
+
+        let path = Path::parse("foo/bar").unwrap();
+        client.delete(&path).await.unwrap();
+
+        let captured = seen.lock();
+        assert_eq!(captured.len(), 1);
+        assert_eq!(captured[0].kind, OperationKind::Delete);
+        assert_eq!(captured[0].backend, "http");
+        assert_eq!(captured[0].location.as_ref(), Some(&path));
+    }
+
+    #[tokio::test]
+    async fn put_populates_operation_extension() {
+        let (client, seen, mock) = setup().await;
+        mock.push(Response::new("".to_string()));
+
+        let path = Path::parse("uploads/x").unwrap();
+        client
+            .put(&path, PutPayload::from_static(b"hi"), Attributes::default())
+            .await
+            .unwrap();
+
+        let captured = seen.lock();
+        assert!(captured.iter().any(|op| op.kind == OperationKind::Put
+            && op.backend == "http"
+            && op.location.as_ref() == Some(&path)));
+    }
+
+    #[tokio::test]
+    async fn get_populates_operation_extension() {
+        let (client, seen, mock) = setup().await;
+        mock.push(Response::new("body".to_string()));
+
+        let path = Path::parse("files/y").unwrap();
+        let mut ctx = RetryContext::new(&RetryConfig::default());
+        client
+            .get_request(&mut ctx, &path, GetOptions::default())
+            .await
+            .unwrap();
+
+        let captured = seen.lock();
+        assert_eq!(captured.len(), 1);
+        assert_eq!(captured[0].kind, OperationKind::Get);
+        assert_eq!(captured[0].backend, "http");
+        assert_eq!(captured[0].location.as_ref(), Some(&path));
+    }
+}