Open source aws_s3 crate (#34631)

GitOrigin-RevId: 8e147733cc785904fbf7a7689fd8d7329131d0dc

Author: emmaling27

Cargo.lock

@@ -18,6 +18,7 @@ csv-async = "1.2"
 atomic_refcell = "0.1.13"
 aws-config = "1.1"
 aws-sdk-s3 = "1.14.0"
+aws-smithy-http = "0.60.8"
 aws-types = "1.1"
 axum = { version = "0.7", features = [ "ws", "original-uri", "macros", "multipart" ] }
 axum-extra = { version = "0.9.3", features = [ "typed-header", "cookie" ] }

Cargo.toml

@@ -0,0 +1,37 @@
+[package]
+name = "aws_s3"
+version = "0.1.0"
+edition = "2021"
+license = "LicenseRef-FSL-1.1-Apache-2.0"
+
+[lib]
+doctest = false
+
+[dependencies]
+anyhow = { workspace = true }
+async-trait = { workspace = true }
+aws-config = { workspace = true }
+aws-sdk-s3 = { workspace = true }
+aws-smithy-http = { workspace = true }
+aws_utils = { path = "../../crates/aws_utils" }
+bytes = { workspace = true }
+common = { path = "../../crates/common" }
+fastrace = { workspace = true }
+futures = { workspace = true }
+http = { workspace = true }
+metrics = { path = "../../crates/metrics" }
+pb = { path = "../../crates/pb" }
+serde_json = { workspace = true }
+storage = { path = "../../crates/storage" }
+tokio = { workspace = true }
+
+[dev-dependencies]
+common = { path = "../../crates/common", features = ["testing"] }
+convex_macro = { path = "../../crates/convex_macro" }
+metrics = { path = "../../crates/metrics", features = ["testing"] }
+runtime = { path = "../../crates/runtime" }
+storage = { path = "../../crates/storage", features = ["testing"] }
+tokio-stream = { workspace = true }
+
+[lints]
+workspace = true

crates/aws_s3/Cargo.toml

@@ -0,0 +1,32 @@
+#![feature(future_join)]
+#![feature(let_chains)]
+#![feature(coroutines)]
+#![feature(iter_from_coroutine)]
+#![feature(iterator_try_collect)]
+
+use aws_sdk_s3::primitives::ByteStream;
+use bytes::Bytes;
+use futures::{
+    Stream,
+    TryStreamExt,
+};
+
+mod metrics;
+pub mod storage;
+mod types;
+
+/// For reasons unknown, the AWS SDK folks have decided that the public APIs
+/// can't contain any third-party types, so the `ByteStream` no longer
+/// implements `Stream`.
+/// We have to yoink this adapter trait from one of their internal crates to
+/// re-add the implementation.
+pub trait ByteStreamCompat {
+    fn into_stream(self) -> impl Stream<Item = Result<Bytes, std::io::Error>>;
+}
+
+impl ByteStreamCompat for ByteStream {
+    fn into_stream(self) -> impl Stream<Item = Result<Bytes, std::io::Error>> {
+        aws_smithy_http::futures_stream_adapter::FuturesStreamCompatByteStream::new(self)
+            .map_err(Into::into)
+    }
+}

crates/aws_s3/src/lib.rs

@@ -0,0 +1,23 @@
+use metrics::{
+    log_distribution,
+    register_convex_histogram,
+    StatusTimer,
+    STATUS_LABEL,
+};
+
+register_convex_histogram!(
+    SIGN_URL_SECONDS,
+    "Time to fetch a presigned S3 link",
+    &STATUS_LABEL
+);
+pub fn sign_url_timer() -> StatusTimer {
+    StatusTimer::new(&SIGN_URL_SECONDS)
+}
+
+register_convex_histogram!(
+    AWS_S3_PART_UPLOAD_SIZE_BYTES,
+    "The size in bytes of each part of an s3 multi part upload.",
+);
+pub fn log_aws_s3_part_upload_size_bytes(size: usize) {
+    log_distribution(&AWS_S3_PART_UPLOAD_SIZE_BYTES, size as f64)
+}