use http scheme; add context for warning logs

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "object-store"
-version = "1.0.0-alpha.6"
+version = "1.0.0-alpha.8"
 authors = ["Stephen Cirner <scirner@figure.com>"]
 edition = "2024"
 
@@ -27,7 +27,6 @@ prost = "0.14"
 prost-types = "0.14"
 quick-error = "2.0"
 rand = "0.9.2"
-reqwest = { version = "0.12" }
 serde = "1.0"
 serde_json = "1.0"
 sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "macros", "uuid", "chrono", "json"] }

src/domain.rs

@@ -1,9 +1,7 @@
 use crate::config::Config;
 use crate::datastore::{AuthType, KeyType, Object, PublicKey};
 use crate::pb::public_key_response::Impl::HeaderAuth as HeaderAuthEnumResponse;
-use crate::pb::{
-    HeaderAuth, ObjectMetadata, ObjectResponse, PublicKeyResponse, Uuid, public_key::Key,
-};
+use crate::pb::{HeaderAuth, ObjectMetadata, ObjectResponse, PublicKeyResponse, public_key::Key};
 use crate::proto_helpers::StringUtil;
 use crate::types::{OsError, Result};
 
@@ -25,12 +23,8 @@ pub trait ObjectApiResponse {
 impl ObjectApiResponse for Object {
     fn to_response(&self, config: &Config) -> Result<ObjectResponse> {
         Ok(ObjectResponse {
-            uuid: Some(Uuid {
-                value: self.uuid.as_hyphenated().to_string(),
-            }),
-            dime_uuid: Some(Uuid {
-                value: self.dime_uuid.as_hyphenated().to_string(),
-            }),
+            uuid: Some(self.uuid.into()),
+            dime_uuid: Some(self.dime_uuid.into()),
             hash: self.hash.decoded()?,
             uri: format!("object://{}/{}", &config.uri_host, &self.hash),
             bucket: config.storage_config.storage_base_path.clone(),
@@ -91,9 +85,7 @@ impl PublicKeyApiResponse for PublicKey {
             None => None,
         };
         let response = PublicKeyResponse {
-            uuid: Some(Uuid {
-                value: self.uuid.as_hyphenated().to_string(),
-            }),
+            uuid: Some(self.uuid.into()),
             public_key: Some(public_key.into()),
             url: self.url,
             r#impl,

src/mailbox.rs

@@ -1,6 +1,6 @@
 use crate::datastore;
 use crate::pb::mailbox_service_server::MailboxService;
-use crate::pb::{AckRequest, GetRequest, MailPayload, Uuid};
+use crate::pb::{AckRequest, GetRequest, MailPayload};
 use crate::proto_helpers::VecUtil;
 use crate::types::{GrpcResult, OsError};
 use crate::{
@@ -79,15 +79,15 @@ impl MailboxService for MailboxGrpc {
             for (mailbox_uuid, object) in results {
                 let payload = match object.payload {
                     Some(payload) => MailPayload {
-                        uuid: Some(Uuid {
-                            value: mailbox_uuid.as_hyphenated().to_string(),
-                        }),
+                        uuid: Some(mailbox_uuid.into()),
                         data: payload,
                     },
                     None => {
                         log::error!(
-                            "mailbox object without a payload {}",
+                            "mailbox object without a payload: uuid={} hash={} dime_length={}",
                             object.uuid.as_hyphenated(),
+                            object.hash,
+                            object.dime_length,
                         );
 
                         if tx

src/middleware/logging_grpc.rs

@@ -59,6 +59,7 @@ where
 
     fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
         let headers = req.headers().clone();
+        let uri = req.uri().clone();
         let trace_header = self.trace_header.clone();
         let lower_logging_bounds = self.lower_logging_bounds;
         let upper_logging_bounds = self.upper_logging_bounds;
@@ -78,11 +79,26 @@ where
             let elapsed_seconds = start.elapsed().as_secs_f64();
 
             if elapsed_seconds > upper_logging_bounds {
-                log::warn!("Trace ID: {} took {} second(s)", trace_id, elapsed_seconds);
+                log::warn!(
+                    "Trace ID: {} uri={} took {:.3}s",
+                    trace_id,
+                    uri,
+                    elapsed_seconds
+                );
             } else if elapsed_seconds > lower_logging_bounds {
-                log::info!("Trace ID: {} took {} second(s)", trace_id, elapsed_seconds);
+                log::info!(
+                    "Trace ID: {} uri={} took {:.3}s",
+                    trace_id,
+                    uri,
+                    elapsed_seconds
+                );
             } else {
-                log::trace!("Trace ID: {} took {} second(s)", trace_id, elapsed_seconds);
+                log::trace!(
+                    "Trace ID: {} uri={} took {:.3}s",
+                    trace_id,
+                    uri,
+                    elapsed_seconds
+                );
             }
 
             Ok(response)

src/middleware/minitrace_grpc.rs

@@ -1,7 +1,6 @@
 use fastrace::prelude::*;
-use http::{Request, Response};
+use http::{HeaderMap, Request, Response};
 use http_body::Body;
-use reqwest::header::HeaderMap;
 use std::{
     fmt::Debug,
     task::{Context, Poll},
@@ -118,15 +117,17 @@ where
             span
         };
 
+        let uri = req.uri().clone();
         let future = self.inner.call(req).in_span(root_span);
 
         Box::pin(async move {
             let response = future.await?;
 
-            match response.status_code(default_status_code) {
+            let status_code = response.status_code(default_status_code);
+            match status_code {
                 tonic::Code::Ok => {}
                 _ => {
-                    log::warn!("rpc call failed");
+                    log::warn!("rpc call failed: uri={} status={:?}", uri, status_code);
                 }
             };
 

src/proto_helpers.rs

@@ -8,7 +8,7 @@ use std::str::FromStr;
 
 use crate::pb::chunk::Impl::{Data, End, Value};
 use crate::pb::chunk_bidi::Impl::{Chunk as ChunkEnum, MultiStreamHeader as MultiStreamHeaderEnum};
-use crate::pb::{Audience, Chunk, ChunkBidi, ChunkEnd, ObjectResponse, StreamHeader};
+use crate::pb::{Audience, Chunk, ChunkBidi, ChunkEnd, ObjectResponse, StreamHeader, Uuid};
 use crate::pb::{MultiStreamHeader, PublicKey, public_key::Key};
 
 pub fn create_multi_stream_header(
@@ -147,6 +147,14 @@ impl From<Vec<u8>> for PublicKey {
     }
 }
 
+impl From<uuid::Uuid> for Uuid {
+    fn from(value: uuid::Uuid) -> Self {
+        Self {
+            value: value.as_hyphenated().to_string(),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use base64::{Engine, prelude::BASE64_STANDARD};

src/replication/mod.rs

@@ -288,16 +288,36 @@ impl ReplicationState {
                 }) => {
                     match error {
                         ReplicationError::ClientCacheError(_) => {
-                            log::error!("Failed replication for {} - {:?}", public_key, error)
+                            log::error!(
+                                "Failed replication for {} to {} - {:?}",
+                                public_key,
+                                url,
+                                error
+                            )
                         }
                         ReplicationError::CrateError(_) => {
-                            log::error!("Failed replication for {} - {:?}", public_key, error)
+                            log::error!(
+                                "Failed replication for {} to {} - {:?}",
+                                public_key,
+                                url,
+                                error
+                            )
                         }
                         ReplicationError::TonicStatusError(_) => {
-                            log::error!("Failed replication for {} - {:?}", public_key, error)
+                            log::error!(
+                                "Failed replication for {} to {} - {:?}",
+                                public_key,
+                                url,
+                                error
+                            )
                         }
                         ReplicationError::TonicTransportError(e) => {
-                            log::trace!("Failed replication for {} - {:?}", public_key, e)
+                            log::trace!(
+                                "Failed replication for {} to {} - {:?}",
+                                public_key,
+                                url,
+                                e
+                            )
                         }
                     }
                     (client, url)

src/storage/file_system.rs

@@ -60,7 +60,7 @@ impl Storage for FileSystem {
     #[trace(name = "file_system::store")]
     async fn store(&self, path: &StoragePath, content_length: usize, data: &[u8]) -> Result<()> {
         if let Err(e) = self.validate_content_length(path, content_length, data) {
-            log::warn!("{:?}", e);
+            log::warn!("store: {:?}", e);
         }
 
         let file_path = self.get_path(path);
@@ -97,7 +97,7 @@ impl Storage for FileSystem {
         })?;
 
         if let Err(e) = self.validate_content_length(path, content_length, &data) {
-            log::warn!("{:?}", e);
+            log::warn!("fetch: {:?}", e);
         }
 
         Ok(data)

src/storage/google_cloud.rs

@@ -56,7 +56,7 @@ impl Storage for GoogleCloud {
     #[trace(name = "google_cloud::store")]
     async fn store(&self, path: &StoragePath, content_length: usize, data: &[u8]) -> Result<()> {
         if let Err(e) = self.validate_content_length(path, content_length, data) {
-            log::warn!("{:?}", e);
+            log::warn!("store: {:?}", e);
         }
 
         let bucket_path = self.bucket();
@@ -113,7 +113,7 @@ impl Storage for GoogleCloud {
         };
 
         if let Err(e) = self.validate_content_length(path, content_length, &data) {
-            log::warn!("{:?}", e);
+            log::warn!("fetch: {:?}", e);
         }
 
         Ok(data)

tests/common/client.rs

@@ -7,19 +7,19 @@ use object_store::pb::{
 use tonic::transport::Channel;
 
 pub async fn get_admin_client(addr: SocketAddr) -> AdminServiceClient<Channel> {
-    AdminServiceClient::connect(format!("tcp://{}", addr))
+    AdminServiceClient::connect(format!("http://{}", addr))
         .await
         .unwrap()
 }
 
 pub async fn get_mailbox_client(addr: SocketAddr) -> MailboxServiceClient<Channel> {
-    MailboxServiceClient::connect(format!("tcp://{}", addr))
+    MailboxServiceClient::connect(format!("http://{}", addr))
         .await
         .unwrap()
 }
 
 pub async fn get_object_client(addr: SocketAddr) -> ObjectServiceClient<Channel> {
-    ObjectServiceClient::connect(format!("tcp://{}", addr))
+    ObjectServiceClient::connect(format!("http://{}", addr))
         .await
         .unwrap()
 }

tests/common/data.rs

@@ -102,7 +102,7 @@ pub fn seed_cache(cache: &mut Cache, remote_config: &Config) {
         ..test_public_key(party_1().0.public_key)
     });
     cache.add_public_key(PublicKey {
-        url: String::from(format!("tcp://{}", remote_config.url)),
+        url: String::from(format!("http://{}", remote_config.url)),
         auth_data: Some(String::from("X-Test-Header:test_value")),
         ..test_public_key(party_2().0.public_key)
     });

tests/object.rs

@@ -35,7 +35,7 @@ fn add_keys_cache(cache: Arc<Mutex<Cache>>) {
         ..test_public_key(party_1().0.public_key)
     });
     guard.add_public_key(PublicKey {
-        url: String::from("tcp://party2:8080"),
+        url: String::from("http://party2:8080"),
         auth_data: Some(String::from("x-test-header:test_value_2")),
         ..test_public_key(party_2().0.public_key)
     });

tests/replication.rs

@@ -38,7 +38,7 @@ async fn client_caching() -> Result<()> {
     let (_, _, replication_state_one, config_one) =
         start_test_server(config_one, Some(&config_two)).await;
 
-    let url_one = format!("tcp://{}", config_one.url);
+    let url_one = format!("http://{}", config_one.url);
 
     let mut client_cache = replication_state_one.client_cache.lock().await;
 
@@ -61,7 +61,7 @@ async fn client_caching() -> Result<()> {
 
     // Client 2:
     let (_, _, _, config_two) = start_test_server(config_two, None).await;
-    let url_two = format!("tcp://{}", config_two.url);
+    let url_two = format!("http://{}", config_two.url);
 
     let client_two = client_cache.request(&url_two).await.unwrap().unwrap();