diff --git a/Cargo.lock b/Cargo.lock index 5168f79..1abaf05 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1920,7 +1920,7 @@ dependencies = [ [[package]] name = "object-store" -version = "1.0.0-alpha.6" +version = "1.0.0-alpha.8" dependencies = [ "async-trait", "base64 0.22.1", @@ -1944,7 +1944,6 @@ dependencies = [ "prost-types", "quick-error", "rand 0.9.2", - "reqwest", "serde", "serde_json", "sqlx", diff --git a/Cargo.toml b/Cargo.toml index 110efa3..1dee28f 100644 --- a/Cargo.toml +++ b/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 "] 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"] } diff --git a/src/domain.rs b/src/domain.rs index 707a03d..01b0ff9 100644 --- a/src/domain.rs +++ b/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 { 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, diff --git a/src/mailbox.rs b/src/mailbox.rs index b3ce12b..3ced665 100644 --- a/src/mailbox.rs +++ b/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 diff --git a/src/middleware/logging_grpc.rs b/src/middleware/logging_grpc.rs index bca8209..e14da6c 100644 --- a/src/middleware/logging_grpc.rs +++ b/src/middleware/logging_grpc.rs @@ -59,6 +59,7 @@ where fn call(&mut self, req: Request) -> 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) diff --git a/src/middleware/minitrace_grpc.rs b/src/middleware/minitrace_grpc.rs index 4a3fb56..bb95ef3 100644 --- a/src/middleware/minitrace_grpc.rs +++ b/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); } }; diff --git a/src/proto_helpers.rs b/src/proto_helpers.rs index 9986fba..3f3c124 100644 --- a/src/proto_helpers.rs +++ b/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> for PublicKey { } } +impl From 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}; diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 6d1d9b5..b6e06ea 100644 --- a/src/replication/mod.rs +++ b/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) diff --git a/src/storage/file_system.rs b/src/storage/file_system.rs index e284b25..7619b43 100644 --- a/src/storage/file_system.rs +++ b/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) diff --git a/src/storage/google_cloud.rs b/src/storage/google_cloud.rs index 1a1d13f..f2dddf5 100644 --- a/src/storage/google_cloud.rs +++ b/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) diff --git a/tests/common/client.rs b/tests/common/client.rs index b516f93..65d3320 100644 --- a/tests/common/client.rs +++ b/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 { - AdminServiceClient::connect(format!("tcp://{}", addr)) + AdminServiceClient::connect(format!("http://{}", addr)) .await .unwrap() } pub async fn get_mailbox_client(addr: SocketAddr) -> MailboxServiceClient { - MailboxServiceClient::connect(format!("tcp://{}", addr)) + MailboxServiceClient::connect(format!("http://{}", addr)) .await .unwrap() } pub async fn get_object_client(addr: SocketAddr) -> ObjectServiceClient { - ObjectServiceClient::connect(format!("tcp://{}", addr)) + ObjectServiceClient::connect(format!("http://{}", addr)) .await .unwrap() } diff --git a/tests/common/data.rs b/tests/common/data.rs index d19846d..84bb1d3 100644 --- a/tests/common/data.rs +++ b/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) }); diff --git a/tests/object.rs b/tests/object.rs index ab0f1f6..9efab98 100644 --- a/tests/object.rs +++ b/tests/object.rs @@ -35,7 +35,7 @@ fn add_keys_cache(cache: Arc>) { ..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) }); diff --git a/tests/replication.rs b/tests/replication.rs index 1cee22c..58e0f6d 100644 --- a/tests/replication.rs +++ b/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();