Add bucket and other info to AWS errors (#1123)

* Add bucket and other info to AWS errors

When you have multiple file_info_pollers running and one of them may be
misconfigured, it is difficult to tell from (usually) the
`ListObjectsV2Error` which poller is failing. 

This PR adds the bucket and file key or prefix for the operation that
was being attempted.

Adding the extra information bumps the Err size to 216 bytes, so we box
it to keep from tripping clippy’s `result_large_err` lint.

* tonic has been upgraded, remove clippy.toml

We no longer have to deal with tonic’s large err variants in file-store.
So we can remove this allowance.

* Box ClientError for large variant

Even though this will go away soon

Author: michaeldjeffrey

clippy.toml

@@ -23,7 +23,7 @@ pub enum Error {
     FileInfo(#[from] FileInfoError),
 
     #[error("aws error: {0}")]
-    Aws(#[from] AwsError),
+    Aws(#[from] Box<AwsError>),
 
     #[error("channel error: {0}")]
     Channel(#[from] ChannelError),
@@ -90,6 +90,8 @@ impl ChannelError {
 }
 
 mod aws_error {
+    use std::path::PathBuf;
+
     use super::Error;
 
     use aws_sdk_s3::error::SdkError;
@@ -103,41 +105,101 @@ mod aws_error {
 
     #[derive(thiserror::Error, Debug)]
     pub enum AwsError {
-        #[error("put_object: {0}")]
-        PutObject(PutObjectError),
+        #[error("put_object ({bucket}::{file_path}): {source}")]
+        PutObject {
+            source: PutObjectError,
+            bucket: String,
+            file_path: PathBuf,
+        },
 
-        #[error("pub_object_byte_stream: {0}")]
-        PubObjectByteStream(ByteStreamError),
+        #[error("pub_object_byte_stream ({bucket}::{file_path}): {source}")]
+        PubObjectByteStream {
+            source: ByteStreamError,
+            bucket: String,
+            file_path: PathBuf,
+        },
 
-        #[error("delete_object: {0}")]
-        DeleteObject(DeleteObjectError),
+        #[error("delete_object ({bucket}::{key}): {source}")]
+        DeleteObject {
+            source: DeleteObjectError,
+            bucket: String,
+            key: String,
+        },
 
-        #[error("get_object: {0}")]
-        GetObject(GetObjectError),
+        #[error("get_object ({bucket}::{key}): {source}")]
+        GetObject {
+            source: GetObjectError,
+            bucket: String,
+            key: String,
+        },
 
-        #[error("list_object: {0}")]
-        ListObject(ListObjectsV2Error),
+        #[error("list_object ({bucket}::{prefix}): {source}")]
+        ListObject {
+            source: ListObjectsV2Error,
+            bucket: String,
+            prefix: String,
+        },
     }
 
     impl AwsError {
-        pub fn put_object_error(err: SdkError<PutObjectError>) -> Error {
-            Error::Aws(Self::PutObject(err.into_service_error()))
+        pub fn put_object_error(
+            err: SdkError<PutObjectError>,
+            bucket: &str,
+            file_path: impl Into<PathBuf>,
+        ) -> Error {
+            Error::Aws(Box::new(Self::PutObject {
+                source: err.into_service_error(),
+                bucket: bucket.into(),
+                file_path: file_path.into(),
+            }))
         }
 
-        pub fn pub_object_byte_stream_error(err: ByteStreamError) -> Error {
-            Error::Aws(Self::PubObjectByteStream(err))
+        pub fn pub_object_byte_stream_error(
+            err: ByteStreamError,
+            bucket: impl Into<String>,
+            file_path: impl Into<PathBuf>,
+        ) -> Error {
+            Error::Aws(Box::new(Self::PubObjectByteStream {
+                source: err,
+                bucket: bucket.into(),
+                file_path: file_path.into(),
+            }))
         }
 
-        pub fn delete_object_error(err: SdkError<DeleteObjectError>) -> Error {
-            Error::Aws(Self::DeleteObject(err.into_service_error()))
+        pub fn delete_object_error(
+            err: SdkError<DeleteObjectError>,
+            bucket: impl Into<String>,
+            key: impl Into<String>,
+        ) -> Error {
+            Error::Aws(Box::new(Self::DeleteObject {
+                source: err.into_service_error(),
+                bucket: bucket.into(),
+                key: key.into(),
+            }))
         }
 
-        pub fn get_object_error(err: SdkError<GetObjectError>) -> Error {
-            Error::Aws(Self::GetObject(err.into_service_error()))
+        pub fn get_object_error(
+            err: SdkError<GetObjectError>,
+            bucket: impl Into<String>,
+            key: impl Into<String>,
+        ) -> Error {
+            Error::Aws(Box::new(Self::GetObject {
+                source: err.into_service_error(),
+                bucket: bucket.into(),
+                key: key.into(),
+            }))
         }
 
-        pub fn list_object_error(err: SdkError<ListObjectsV2Error>) -> Error {
-            Error::Aws(Self::ListObject(err.into_service_error()))
+        pub fn list_object_error(
+            err: SdkError<ListObjectsV2Error>,
+            bucket: &str,
+            prefix: &str,
+        ) -> Error {
+            Error::Aws(Box::new(Self::ListObject {
+                source: err.into_service_error(),
+                bucket: bucket.to_string(),
+                prefix: prefix.to_string(),
+            }))
         }
     }
 }

file_store/src/error.rs

@@ -134,17 +134,18 @@ fn _list_files(
     after: Option<DateTime<Utc>>,
 ) -> FileInfoStream {
     let prefix = prefix.into();
+    let bucket = bucket.into();
 
     client
         .list_objects_v2()
-        .bucket(bucket)
+        .bucket(&bucket)
         .prefix(&prefix)
         .set_start_after(after.map(|dt| FileInfo::from_maybe_dotted_prefix(&prefix, dt).into()))
         .into_paginator()
         .send()
         .into_stream_03x()
         .map_ok(|page| stream::iter(page.contents.unwrap_or_default()).map(Ok))
-        .map_err(AwsError::list_object_error)
+        .map_err(move |err| AwsError::list_object_error(err, &bucket, &prefix))
         .try_flatten()
         .try_filter_map(|file| {
             future::ready(FileInfo::try_from(&file).map(Some).map_err(Error::from))
@@ -169,21 +170,23 @@ where
 }
 
 pub async fn put_file(client: &Client, bucket: impl Into<String>, file: &Path) -> Result {
+    let bucket: String = bucket.into();
+
     let byte_stream = ByteStream::from_path(&file)
-        .map_err(AwsError::pub_object_byte_stream_error)
+        .map_err(|err| AwsError::pub_object_byte_stream_error(err, &bucket, file))
         .await?;
 
     poc_metrics::record_duration!(
         "file_store_put_duration",
         client
             .put_object()
-            .bucket(bucket)
+            .bucket(&bucket)
             .key(file.file_name().map(|name| name.to_string_lossy()).unwrap())
             .body(byte_stream)
             .content_type("application/octet-stream")
             .send()
             .map_ok(|_| ())
-            .map_err(AwsError::put_object_error)
+            .map_err(|err| AwsError::put_object_error(err, &bucket, file))
             .await
     )
 }
@@ -193,15 +196,18 @@ pub async fn remove_file(
     bucket: impl Into<String>,
     key: impl Into<String>,
 ) -> Result {
+    let bucket = bucket.into();
+    let key = key.into();
+
     poc_metrics::record_duration!(
         "file_store_remove_duration",
         client
             .delete_object()
-            .bucket(bucket)
-            .key(key)
+            .bucket(&bucket)
+            .key(&key)
             .send()
             .map_ok(|_| ())
-            .map_err(AwsError::delete_object_error)
+            .map_err(|err| AwsError::delete_object_error(err, bucket, key))
             .await
     )
 }
@@ -288,13 +294,16 @@ async fn get_byte_stream(
     bucket: impl Into<String>,
     key: impl Into<String>,
 ) -> Result<ByteStream> {
+    let bucket = bucket.into();
+    let key = key.into();
+
     client
         .get_object()
-        .bucket(bucket)
-        .key(key)
+        .bucket(&bucket)
+        .key(&key)
         .send()
         .map_ok(|output| output.body)
-        .map_err(AwsError::get_object_error)
+        .map_err(|err| AwsError::get_object_error(err, bucket, key))
         .fuse()
         .await
 }

file_store/src/lib.rs

@@ -26,7 +26,13 @@ pub enum RegionCacheError {
     #[error("region not found: {0}")]
     RegionNotFound(ProtoRegion),
     #[error("error querying gateway api")]
-    GatewayApiError(ClientError),
+    GatewayApiError(Box<ClientError>),
+}
+
+impl From<ClientError> for RegionCacheError {
+    fn from(err: ClientError) -> Self {
+        Self::GatewayApiError(Box::new(err))
+    }
 }
 
 impl<G> RegionCache<G>
@@ -62,7 +68,7 @@ where
                         .await;
                     Ok(res)
                 }
-                Err(err) => Err(RegionCacheError::GatewayApiError(err)),
+                Err(err) => Err(err)?,
             },
         }
     }