Decouple ObjectStore from Reqwest (#7183)

* Create HttpClient

* WIP

* Decouple from reqwest

* Hook up HttpConnector

* WIP

* Update other stores

* Format

* Remove out of date comment

* RAT

* Test fixes

* Lints

* Fix feature flags

Author: tustvold

object_store/Cargo.toml

@@ -45,24 +45,28 @@ walkdir = { version = "2", optional = true }
 
 # Cloud storage support
 base64 = { version = "0.22", default-features = false, features = ["std"], optional = true }
+form_urlencoded = { version = "1.2", optional = true }
+http = { version = "1.2.0", optional = true }
+http-body-util = { version = "0.1", optional = true }
+httparse = { version = "1.8.0", default-features = false, features = ["std"], optional = true }
 hyper = { version = "1.2", default-features = false, optional = true }
+md-5 = { version = "0.10.6", default-features = false, optional = true }
 quick-xml = { version = "0.37.0", features = ["serialize", "overlapped-lists"], optional = true }
-serde = { version = "1.0", default-features = false, features = ["derive"], optional = true }
-serde_json = { version = "1.0", default-features = false, optional = true }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"], optional = true }
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-native-roots", "http2"], optional = true }
 ring = { version = "0.17", default-features = false, features = ["std"], optional = true }
 rustls-pemfile = { version = "2.0", default-features = false, features = ["std"], optional = true }
+serde = { version = "1.0", default-features = false, features = ["derive"], optional = true }
+serde_json = { version = "1.0", default-features = false, features = ["std"], optional = true }
+serde_urlencoded = { version = "0.7", optional = true }
 tokio = { version = "1.29.0", features = ["sync", "macros", "rt", "time", "io-util"] }
-md-5 = { version = "0.10.6", default-features = false, optional = true }
-httparse = { version = "1.8.0", default-features = false, features = ["std"], optional = true }
 
 [target.'cfg(target_family="unix")'.dev-dependencies]
 nix = { version = "0.29.0", features = ["fs"] }
 
 [features]
 default = ["fs"]
-cloud = ["serde", "serde_json", "quick-xml", "hyper", "reqwest", "reqwest/json", "reqwest/stream", "chrono/serde", "base64", "rand", "ring"]
+cloud = ["serde", "serde_json", "quick-xml", "hyper", "reqwest", "reqwest/stream", "chrono/serde", "base64", "rand", "ring", "dep:http", "http-body-util", "form_urlencoded", "serde_urlencoded"]
 azure = ["cloud", "httparse"]
 fs = ["walkdir"]
 gcp = ["cloud", "rustls-pemfile"]
@@ -72,16 +76,13 @@ tls-webpki-roots = ["reqwest?/rustls-tls-webpki-roots"]
 integration = []
 
 [dev-dependencies] # In alphabetical order
-futures-test = "0.3"
 hyper = { version = "1.2", features = ["server"] }
 hyper-util = "0.1"
-http-body-util = "0.1"
 rand = "0.8"
 tempfile = "3.1.0"
 regex = "1.11.1"
 # The "gzip" feature for reqwest is enabled for an integration test.
 reqwest = { version = "0.12", features = ["gzip"] }
-http = "1.1.0"
 
 [[test]]
 name = "get_range_file"

object_store/src/aws/builder.rs

@@ -23,7 +23,7 @@ use crate::aws::{
     AmazonS3, AwsCredential, AwsCredentialProvider, Checksum, S3ConditionalPut, S3CopyIfNotExists,
     STORE,
 };
-use crate::client::TokenCredentialProvider;
+use crate::client::{HttpConnector, ReqwestConnector, TokenCredentialProvider};
 use crate::config::ConfigValue;
 use crate::{ClientConfigKey, ClientOptions, Result, RetryConfig, StaticCredentialProvider};
 use base64::prelude::BASE64_STANDARD;
@@ -171,6 +171,8 @@ pub struct AmazonS3Builder {
     encryption_customer_key_base64: Option<String>,
     /// When set to true, charge requester for bucket operations
     request_payer: ConfigValue<bool>,
+    /// The [`HttpConnector`] to use
+    http_connector: Option<Arc<dyn HttpConnector>>,
 }
 
 /// Configuration keys for [`AmazonS3Builder`]
@@ -882,13 +884,23 @@ impl AmazonS3Builder {
         self
     }
 
+    /// Overrides the [`HttpConnector`], by default uses [`ReqwestConnector`]
+    pub fn with_http_connector<C: HttpConnector>(mut self, connector: C) -> Self {
+        self.http_connector = Some(Arc::new(connector));
+        self
+    }
+
     /// Create a [`AmazonS3`] instance from the provided values,
     /// consuming `self`.
     pub fn build(mut self) -> Result<AmazonS3> {
         if let Some(url) = self.url.take() {
             self.parse_url(&url)?;
         }
 
+        let http = self
+            .http_connector
+            .unwrap_or_else(|| Arc::new(ReqwestConnector::default()));
+
         let bucket = self.bucket_name.ok_or(Error::MissingBucketName)?;
         let region = self.region.unwrap_or_else(|| "us-east-1".to_string());
         let checksum = self.checksum_algorithm.map(|x| x.get()).transpose()?;
@@ -925,11 +937,7 @@ impl AmazonS3Builder {
             let endpoint = format!("https://sts.{region}.amazonaws.com");
 
             // Disallow non-HTTPs requests
-            let client = self
-                .client_options
-                .clone()
-                .with_allow_http(false)
-                .client()?;
+            let options = self.client_options.clone().with_allow_http(false);
 
             let token = WebIdentityProvider {
                 token_path,
@@ -940,16 +948,19 @@ impl AmazonS3Builder {
 
             Arc::new(TokenCredentialProvider::new(
                 token,
-                client,
+                http.connect(&options)?,
                 self.retry_config.clone(),
             )) as _
         } else if let Some(uri) = self.container_credentials_relative_uri {
             info!("Using Task credential provider");
+
+            let options = self.client_options.clone().with_allow_http(true);
+
             Arc::new(TaskCredentialProvider {
                 url: format!("http://169.254.170.2{uri}"),
                 retry: self.retry_config.clone(),
                 // The instance metadata endpoint is access over HTTP
-                client: self.client_options.clone().with_allow_http(true).client()?,
+                client: http.connect(&options)?,
                 cache: Default::default(),
             }) as _
         } else {
@@ -964,7 +975,7 @@ impl AmazonS3Builder {
 
             Arc::new(TokenCredentialProvider::new(
                 token,
-                self.client_options.metadata_client()?,
+                http.connect(&self.client_options.metadata_options())?,
                 self.retry_config.clone(),
             )) as _
         };
@@ -986,7 +997,7 @@ impl AmazonS3Builder {
                             region: region.clone(),
                             credentials: Arc::clone(&credentials),
                         },
-                        self.client_options.client()?,
+                        http.connect(&self.client_options)?,
                         self.retry_config.clone(),
                     )
                     .with_min_ttl(Duration::from_secs(60)), // Credentials only valid for 5 minutes
@@ -1039,7 +1050,8 @@ impl AmazonS3Builder {
             request_payer: self.request_payer.get()?,
         };
 
-        let client = Arc::new(S3Client::new(config)?);
+        let http_client = http.connect(&config.client_options)?;
+        let client = Arc::new(S3Client::new(config, http_client));
 
         Ok(AmazonS3 { client })
     }

object_store/src/aws/client.rs

@@ -22,6 +22,7 @@ use crate::aws::{
     AwsAuthorizer, AwsCredentialProvider, S3ConditionalPut, S3CopyIfNotExists, COPY_SOURCE_HEADER,
     STORE, STRICT_PATH_ENCODE_SET, TAGS_HEADER,
 };
+use crate::client::builder::{HttpRequestBuilder, RequestBuilderError};
 use crate::client::get::GetClient;
 use crate::client::header::{get_etag, HeaderConfig};
 use crate::client::header::{get_put_result, get_version};
@@ -31,7 +32,7 @@ use crate::client::s3::{
     CompleteMultipartUpload, CompleteMultipartUploadResult, CopyPartResult,
     InitiateMultipartUploadResult, ListResponse, PartMetadata,
 };
-use crate::client::GetOptionsExt;
+use crate::client::{GetOptionsExt, HttpClient, HttpError, HttpResponse};
 use crate::multipart::PartId;
 use crate::path::DELIMITER;
 use crate::{
@@ -42,17 +43,15 @@ use async_trait::async_trait;
 use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
 use bytes::{Buf, Bytes};
-use hyper::header::{
+use http::header::{
     CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_LENGTH,
     CONTENT_TYPE,
 };
-use hyper::http::HeaderName;
-use hyper::{http, HeaderMap};
+use http::{HeaderMap, HeaderName, Method};
 use itertools::Itertools;
 use md5::{Digest, Md5};
 use percent_encoding::{utf8_percent_encode, PercentEncode};
 use quick_xml::events::{self as xml_events};
-use reqwest::{Client as ReqwestClient, Method, RequestBuilder, Response};
 use ring::digest;
 use ring::digest::Context;
 use serde::{Deserialize, Serialize};
@@ -67,7 +66,9 @@ const ALGORITHM: &str = "x-amz-checksum-algorithm";
 #[derive(Debug, thiserror::Error)]
 pub(crate) enum Error {
     #[error("Error performing DeleteObjects request: {}", source)]
-    DeleteObjectsRequest { source: crate::client::retry::Error },
+    DeleteObjectsRequest {
+        source: crate::client::retry::RetryError,
+    },
 
     #[error(
         "DeleteObjects request failed for key {}: {} (code: {})",
@@ -82,30 +83,32 @@ pub(crate) enum Error {
     },
 
     #[error("Error getting DeleteObjects response body: {}", source)]
-    DeleteObjectsResponse { source: reqwest::Error },
+    DeleteObjectsResponse { source: HttpError },
 
     #[error("Got invalid DeleteObjects response: {}", source)]
     InvalidDeleteObjectsResponse {
         source: Box<dyn std::error::Error + Send + Sync + 'static>,
     },
 
     #[error("Error performing list request: {}", source)]
-    ListRequest { source: crate::client::retry::Error },
+    ListRequest {
+        source: crate::client::retry::RetryError,
+    },
 
     #[error("Error getting list response body: {}", source)]
-    ListResponseBody { source: reqwest::Error },
+    ListResponseBody { source: HttpError },
 
     #[error("Error getting create multipart response body: {}", source)]
-    CreateMultipartResponseBody { source: reqwest::Error },
+    CreateMultipartResponseBody { source: HttpError },
 
     #[error("Error performing complete multipart request: {}: {}", path, source)]
     CompleteMultipartRequest {
-        source: crate::client::retry::Error,
+        source: crate::client::retry::RetryError,
         path: String,
     },
 
     #[error("Error getting complete multipart response body: {}", source)]
-    CompleteMultipartResponseBody { source: reqwest::Error },
+    CompleteMultipartResponseBody { source: HttpError },
 
     #[error("Got invalid list response: {}", source)]
     InvalidListResponse { source: quick_xml::de::DeError },
@@ -272,7 +275,7 @@ pub enum RequestError {
 
     #[error("Retry")]
     Retry {
-        source: crate::client::retry::Error,
+        source: crate::client::retry::RetryError,
         path: String,
     },
 }
@@ -290,7 +293,7 @@ impl From<RequestError> for crate::Error {
 pub(crate) struct Request<'a> {
     path: &'a Path,
     config: &'a S3Config,
-    builder: RequestBuilder,
+    builder: HttpRequestBuilder,
     payload_sha256: Option<digest::Digest>,
     payload: Option<PutPayload>,
     use_session_creds: bool,
@@ -307,8 +310,8 @@ impl Request<'_> {
 
     pub(crate) fn header<K>(self, k: K, v: &str) -> Self
     where
-        HeaderName: TryFrom<K>,
-        <HeaderName as TryFrom<K>>::Error: Into<http::Error>,
+        K: TryInto<HeaderName>,
+        K::Error: Into<RequestBuilderError>,
     {
         let builder = self.builder.header(k, v);
         Self { builder, ..self }
@@ -408,7 +411,7 @@ impl Request<'_> {
         self
     }
 
-    pub(crate) async fn send(self) -> Result<Response, RequestError> {
+    pub(crate) async fn send(self) -> Result<HttpResponse, RequestError> {
         let credential = match self.use_session_creds {
             true => self.config.get_session_credential().await?,
             false => SessionCredential {
@@ -446,13 +449,12 @@ impl Request<'_> {
 #[derive(Debug)]
 pub(crate) struct S3Client {
     pub config: S3Config,
-    pub client: ReqwestClient,
+    pub client: HttpClient,
 }
 
 impl S3Client {
-    pub(crate) fn new(config: S3Config) -> Result<Self> {
-        let client = config.client_options.client()?;
-        Ok(Self { config, client })
+    pub(crate) fn new(config: S3Config, client: HttpClient) -> Self {
+        Self { config, client }
     }
 
     pub(crate) fn request<'a>(&'a self, method: Method, path: &'a Path) -> Request<'a> {
@@ -544,6 +546,7 @@ impl S3Client {
             .send_retry(&self.config.retry_config)
             .await
             .map_err(|source| Error::DeleteObjectsRequest { source })?
+            .into_body()
             .bytes()
             .await
             .map_err(|source| Error::DeleteObjectsResponse { source })?;
@@ -641,6 +644,7 @@ impl S3Client {
             .idempotent(true)
             .send()
             .await?
+            .into_body()
             .bytes()
             .await
             .map_err(|source| Error::CreateMultipartResponseBody { source })?;
@@ -683,17 +687,17 @@ impl S3Client {
             // If SSE-C is used, we must include the encryption headers in every upload request.
             request = request.with_encryption_headers();
         }
-        let response = request.send().await?;
-        let checksum_sha256 = response
-            .headers()
+        let (parts, body) = request.send().await?.into_parts();
+        let checksum_sha256 = parts
+            .headers
             .get(SHA256_CHECKSUM)
             .and_then(|v| v.to_str().ok())
             .map(|v| v.to_string());
 
         let e_tag = match is_copy {
-            false => get_etag(response.headers()).map_err(|source| Error::Metadata { source })?,
+            false => get_etag(&parts.headers).map_err(|source| Error::Metadata { source })?,
             true => {
-                let response = response
+                let response = body
                     .bytes()
                     .await
                     .map_err(|source| Error::CreateMultipartResponseBody { source })?;
@@ -756,7 +760,7 @@ impl S3Client {
 
         let request = self
             .client
-            .request(Method::POST, url)
+            .post(url)
             .query(&[("uploadId", upload_id)])
             .body(body)
             .with_aws_sigv4(credential.authorizer(), None);
@@ -781,6 +785,7 @@ impl S3Client {
             .map_err(|source| Error::Metadata { source })?;
 
         let data = response
+            .into_body()
             .bytes()
             .await
             .map_err(|source| Error::CompleteMultipartResponseBody { source })?;
@@ -795,7 +800,7 @@ impl S3Client {
     }
 
     #[cfg(test)]
-    pub(crate) async fn get_object_tagging(&self, path: &Path) -> Result<Response> {
+    pub(crate) async fn get_object_tagging(&self, path: &Path) -> Result<HttpResponse> {
         let credential = self.config.get_session_credential().await?;
         let url = format!("{}?tagging", self.config.path_url(path));
         let response = self
@@ -821,7 +826,7 @@ impl GetClient for S3Client {
     };
 
     /// Make an S3 GET request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html>
-    async fn get_request(&self, path: &Path, options: GetOptions) -> Result<Response> {
+    async fn get_request(&self, path: &Path, options: GetOptions) -> Result<HttpResponse> {
         let credential = self.config.get_session_credential().await?;
         let url = self.config.path_url(path);
         let method = match options.head {
@@ -895,6 +900,7 @@ impl ListClient for Arc<S3Client> {
             .send_retry(&self.config.retry_config)
             .await
             .map_err(|source| Error::ListRequest { source })?
+            .into_body()
             .bytes()
             .await
             .map_err(|source| Error::ListResponseBody { source })?;