You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2023/04/04 16:01:11 UTC
[incubator-opendal] branch main updated: refactor: Use reqwest blocking client instead of ureq (#1853)
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
The following commit(s) were added to refs/heads/main by this push:
new ee74e181 refactor: Use reqwest blocking client instead of ureq (#1853)
ee74e181 is described below
commit ee74e181e9d93c3ca7b99b25bc0956ec7289552d
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Apr 5 00:01:04 2023 +0800
refactor: Use reqwest blocking client instead of ureq (#1853)
* refactor: Use reqwest blocking client instead of ureq
Signed-off-by: Xuanwo <gi...@xuanwo.io>
* Fix region
Signed-off-by: Xuanwo <gi...@xuanwo.io>
* Configure ureq
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---------
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
.github/workflows/service_test_s3.yml | 3 +
core/Cargo.toml | 1 +
core/src/raw/http_util/body.rs | 55 ++++----
core/src/raw/http_util/client.rs | 237 +++++++++++++++++-----------------
core/src/raw/http_util/dns.rs | 237 ----------------------------------
core/src/raw/http_util/mod.rs | 2 -
core/src/services/s3/backend.rs | 146 +--------------------
7 files changed, 152 insertions(+), 529 deletions(-)
diff --git a/.github/workflows/service_test_s3.yml b/.github/workflows/service_test_s3.yml
index c379b836..72c7a9cd 100644
--- a/.github/workflows/service_test_s3.yml
+++ b/.github/workflows/service_test_s3.yml
@@ -56,6 +56,7 @@ jobs:
OPENDAL_S3_ENDPOINT: ${{ secrets.OPENDAL_S3_ENDPOINT }}
OPENDAL_S3_ACCESS_KEY_ID: ${{ secrets.OPENDAL_S3_ACCESS_KEY_ID }}
OPENDAL_S3_SECRET_ACCESS_KEY: ${{ secrets.OPENDAL_S3_SECRET_ACCESS_KEY }}
+ OPENDAL_S3_REGION: ap-northeast-1
aws_s3_with_virtual_host:
runs-on: ubuntu-latest
@@ -77,6 +78,7 @@ jobs:
OPENDAL_S3_ACCESS_KEY_ID: ${{ secrets.OPENDAL_S3_ACCESS_KEY_ID }}
OPENDAL_S3_SECRET_ACCESS_KEY: ${{ secrets.OPENDAL_S3_SECRET_ACCESS_KEY }}
OPENDAL_S3_ENABLE_VIRTUAL_HOST_STYLE: on
+ OPENDAL_S3_REGION: ap-northeast-1
aws_s3_with_sse_c:
runs-on: ubuntu-latest
@@ -100,6 +102,7 @@ jobs:
OPENDAL_S3_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM: AES256
OPENDAL_S3_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY: MDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDA=
OPENDAL_S3_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5: zZ5FnqcIqUjVwvWmyog4zw==
+ OPENDAL_S3_REGION: ap-northeast-1
minio_s3:
runs-on: ubuntu-latest
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 4b57fbaa..98b97ca8 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -121,6 +121,7 @@ reqsign = "0.8.5"
reqwest = { version = "0.11.13", features = [
"multipart",
"stream",
+ "blocking"
], default-features = false }
rocksdb = { version = "0.20.1", default-features = false, optional = true }
serde = { version = "1", features = ["derive"] }
diff --git a/core/src/raw/http_util/body.rs b/core/src/raw/http_util/body.rs
index 324871d2..ddda0fc4 100644
--- a/core/src/raw/http_util/body.rs
+++ b/core/src/raw/http_util/body.rs
@@ -18,8 +18,6 @@
use std::cmp::min;
use std::cmp::Ordering;
use std::io;
-use std::io::Read;
-use std::io::Write;
use std::task::Context;
use std::task::Poll;
@@ -41,8 +39,6 @@ pub enum Body {
Empty,
/// Body with bytes.
Bytes(Bytes),
- /// Body with a Reader.
- Reader(Box<dyn Read + Send>),
}
impl Default for Body {
@@ -54,35 +50,44 @@ impl Default for Body {
impl Body {
/// Consume the entire body.
pub fn consume(self) -> Result<()> {
- if let Body::Reader(mut r) = self {
- std::io::copy(&mut r, &mut std::io::sink()).map_err(|err| {
- Error::new(ErrorKind::Unexpected, "consuming response")
- .with_operation("http_util::Body::consume")
- .set_source(err)
- })?;
- }
-
Ok(())
}
}
-impl Read for Body {
- fn read(&mut self, mut buf: &mut [u8]) -> io::Result<usize> {
- match self {
- Body::Empty => Ok(0),
- Body::Bytes(bs) => {
- let size = min(bs.len(), buf.len());
- let rbs = bs.split_to(size);
- bs.advance(size);
-
- buf.write_all(&rbs).expect("write all must succeed");
- Ok(size)
- }
- Body::Reader(r) => r.read(buf),
+impl From<Body> for reqwest::blocking::Body {
+ fn from(v: Body) -> Self {
+ match v {
+ Body::Empty => reqwest::blocking::Body::from(""),
+ Body::Bytes(bs) => reqwest::blocking::Body::from(bs),
}
}
}
+/// IncomingBody carries the content returned by remote servers.
+///
+/// # Notes
+///
+/// Client SHOULD NEVER construct this body.
+pub struct IncomingBody {
+ /// # TODO
+ ///
+ /// hyper returns `impl Stream<Item = crate::Result<Bytes>>` but we can't
+ /// write the types in stable. So we will box here.
+ ///
+ /// After [TAIT](https://rust-lang.github.io/rfcs/2515-type_alias_impl_trait.html)
+ /// has been stable, we can change `IncomingAsyncBody` into `IncomingAsyncBody<S>`.
+ #[allow(unused)]
+ inner: reqwest::blocking::Response,
+ #[allow(unused)]
+ size: Option<u64>,
+}
+
+impl IncomingBody {
+ pub fn new(resp: reqwest::blocking::Response, size: Option<u64>) -> Self {
+ IncomingBody { inner: resp, size }
+ }
+}
+
/// Body used in async HTTP requests.
pub enum AsyncBody {
/// An empty body.
diff --git a/core/src/raw/http_util/client.rs b/core/src/raw/http_util/client.rs
index b53a0d56..8bd2c318 100644
--- a/core/src/raw/http_util/client.rs
+++ b/core/src/raw/http_util/client.rs
@@ -15,22 +15,19 @@
// specific language governing permissions and limitations
// under the License.
-use std::env;
use std::fmt::Debug;
use std::fmt::Formatter;
use std::str::FromStr;
-use std::sync::Arc;
+use std::thread;
use futures::TryStreamExt;
use http::Request;
use http::Response;
-use log::debug;
use reqwest::redirect::Policy;
-use reqwest::ClientBuilder;
use reqwest::Url;
use super::body::IncomingAsyncBody;
-use super::dns::*;
+use super::body::IncomingBody;
use super::parse_content_length;
use super::AsyncBody;
use super::Body;
@@ -42,7 +39,7 @@ use crate::Result;
#[derive(Clone)]
pub struct HttpClient {
async_client: reqwest::Client,
- sync_client: ureq::Agent,
+ blocking_client: reqwest::blocking::Client,
}
/// We don't want users to know details about our clients.
@@ -53,148 +50,144 @@ impl Debug for HttpClient {
}
impl HttpClient {
- /// Create a new http client.
+ /// Create a new http client in async context.
pub fn new() -> Result<Self> {
- let async_client = {
- let mut builder = ClientBuilder::new();
-
- // Make sure we don't enable auto gzip decompress.
- builder = builder.no_gzip();
- // Make sure we don't enable auto brotli decompress.
- builder = builder.no_brotli();
- // Make sure we don't enable auto deflate decompress.
- builder = builder.no_deflate();
- // Redirect will be handled by ourselves.
- builder = builder.redirect(Policy::none());
-
- #[cfg(feature = "trust-dns")]
- let builder = builder.dns_resolver(Arc::new(AsyncTrustDnsResolver::new().unwrap()));
- #[cfg(not(feature = "trust-dns"))]
- let builder = builder.dns_resolver(Arc::new(AsyncStdDnsResolver::default()));
-
- builder.build().map_err(|err| {
- Error::new(ErrorKind::Unexpected, "async client build failed").set_source(err)
- })?
- };
-
- let sync_client = {
- let mut builder = ureq::AgentBuilder::new();
-
- for key in ["http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY"] {
- if let Ok(proxy) = env::var(key) {
- // Ignore proxy setting if proxy is invalid.
- if let Ok(proxy) = ureq::Proxy::new(proxy) {
- debug!("sync client: set proxy to {proxy:?}");
- builder = builder.proxy(proxy);
- }
- }
- }
-
- let builder = builder.resolver(StdDnsResolver::default());
-
- builder.build()
- };
+ Self::build(
+ reqwest::ClientBuilder::new(),
+ reqwest::blocking::ClientBuilder::new(),
+ )
+ }
+ /// Build a new http client in async context.
+ pub fn build(
+ async_builder: reqwest::ClientBuilder,
+ blocking_builder: reqwest::blocking::ClientBuilder,
+ ) -> Result<Self> {
Ok(HttpClient {
- async_client,
- sync_client,
+ async_client: Self::build_async_client(async_builder)?,
+ blocking_client: Self::build_blocking_client(blocking_builder)?,
+ })
+ }
+
+ /// Build a new blocking client with given builder.
+ fn build_async_client(mut builder: reqwest::ClientBuilder) -> Result<reqwest::Client> {
+ // Make sure we don't enable auto gzip decompress.
+ builder = builder.no_gzip();
+ // Make sure we don't enable auto brotli decompress.
+ builder = builder.no_brotli();
+ // Make sure we don't enable auto deflate decompress.
+ builder = builder.no_deflate();
+ // Redirect will be handled by ourselves.
+ builder = builder.redirect(Policy::none());
+
+ #[cfg(feature = "trust-dns")]
+ let builder = builder.trust_dns(true);
+
+ builder.build().map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "async client build failed").set_source(err)
})
}
- /// Build a new http client from already built clients.
+ /// Build a new blocking client with given builder.
///
/// # Notes
///
- /// By using this method, it's caller's duty to make sure everything
- /// configured correctly. Like proxy, dns resolver and so on.
+ /// `reqwest::blocking::ClientBuilder::build` will panic if called
+ /// inside async context. So we need to spawn a thread to build it.
///
- /// And this API is an internal API, OpenDAL could change it while bumping
- /// minor versions.
- ///
- /// ## Reminders
- /// ### no auto redirect
- /// OpenDAL will handle all HTTP responses, including redirections.
- /// Auto redirect may cause OpenDAL to fail.
- ///
- /// For reqwest client, please make sure your client's redirect policy is `Policy::none()`.
- /// ```no_run
- /// # use anyhow::Result;
- /// # use reqwest::redirect::Policy;
- /// # fn main() -> Result<()> {
- /// let _client = reqwest::ClientBuilder::new()
- /// .redirect(Policy::none())
- /// .build()?;
- /// # Ok(())
- /// # }
- /// ```
- /// For ureq client, please make sure your client's redirect count is `0`:
- /// ```no_run
- /// # fn main() {
- /// let _client = ureq::AgentBuilder::new().redirects(0).build();
- /// # }
- /// ```
- pub fn with_client(async_client: reqwest::Client, sync_client: ureq::Agent) -> Self {
- Self {
- async_client,
- sync_client,
- }
+ /// And users SHOULD never call blocking clients in async context.
+ fn build_blocking_client(
+ mut builder: reqwest::blocking::ClientBuilder,
+ ) -> Result<reqwest::blocking::Client> {
+ // Make sure we don't enable auto gzip decompress.
+ builder = builder.no_gzip();
+ // Make sure we don't enable auto brotli decompress.
+ builder = builder.no_brotli();
+ // Make sure we don't enable auto deflate decompress.
+ builder = builder.no_deflate();
+ // Redirect will be handled by ourselves.
+ builder = builder.redirect(Policy::none());
+
+ #[cfg(feature = "trust-dns")]
+ let builder = builder.trust_dns(true);
+
+ thread::spawn(|| {
+ builder.build().map_err(|err| {
+ Error::new(ErrorKind::Unexpected, "blocking client build failed").set_source(err)
+ })
+ })
+ .join()
+ .expect("the thread of building blocking client join failed")
}
/// Get the async client from http client.
- pub fn async_client(&self) -> reqwest::Client {
+ pub async fn async_client(&self) -> reqwest::Client {
self.async_client.clone()
}
- /// Get the sync client from http client.
- pub fn sync_client(&self) -> ureq::Agent {
- self.sync_client.clone()
+ /// Get the blocking client from http client.
+ pub fn blocking_client(&self) -> reqwest::blocking::Client {
+ self.blocking_client.clone()
}
/// Send a request in blocking way.
- pub fn send(&self, req: Request<Body>) -> Result<Response<Body>> {
+ pub fn send(&self, req: Request<Body>) -> Result<Response<IncomingBody>> {
+ let is_head = req.method() == http::Method::HEAD;
let (parts, body) = req.into_parts();
- let mut ur = self
- .sync_client
- .request(parts.method.as_str(), &parts.uri.to_string());
- for (k, v) in parts.headers.iter() {
- ur = ur.set(k.as_str(), v.to_str().expect("must be valid header"));
- }
+ let mut req_builder = self
+ .blocking_client
+ .request(
+ parts.method,
+ Url::from_str(&parts.uri.to_string()).expect("input request url must be valid"),
+ )
+ .version(parts.version)
+ .headers(parts.headers);
- let resp = match ur.send(body) {
- Ok(resp) => resp,
- Err(err_resp) => match err_resp {
- ureq::Error::Status(_code, resp) => resp,
- ureq::Error::Transport(transport) => {
- let is_temporary = matches!(
- transport.kind(),
- ureq::ErrorKind::Dns
- | ureq::ErrorKind::ConnectionFailed
- | ureq::ErrorKind::Io
- );
-
- let mut err = Error::new(ErrorKind::Unexpected, "send blocking request")
- .with_operation("http_util::Client::send")
- .set_source(transport);
- if is_temporary {
- err = err.set_temporary();
- }
-
- return Err(err);
- }
- },
- };
+ req_builder = req_builder.body(body);
+
+ let resp = req_builder.send().map_err(|err| {
+ let is_temporary = !(
+ // Builder related error should not be retried.
+ err.is_builder() ||
+ // Error returned by RedirectPolicy.
+ //
+ // We don't set this by hand, just don't allow retry.
+ err.is_redirect() ||
+ // We never use `Response::error_for_status`, just don't allow retry.
+ //
+ // Status should be checked by our services.
+ err.is_status()
+ );
- let mut hr = Response::builder().status(resp.status());
- for name in resp.headers_names() {
- if let Some(value) = resp.header(&name) {
- hr = hr.header(name, value);
+ let mut oerr = Error::new(ErrorKind::Unexpected, "send blocking request")
+ .with_operation("http_util::Client::send")
+ .set_source(err);
+ if is_temporary {
+ oerr = oerr.set_temporary();
}
+
+ oerr
+ })?;
+
+ // Get content length from header so that we can check it.
+ // If the request method is HEAD, we will ignore this.
+ let content_length = if is_head {
+ None
+ } else {
+ parse_content_length(resp.headers()).expect("response content length must be valid")
+ };
+
+ let mut hr = Response::builder()
+ .version(resp.version())
+ .status(resp.status());
+ for (k, v) in resp.headers().iter() {
+ hr = hr.header(k, v);
}
- let resp = hr
- .body(Body::Reader(Box::new(resp.into_reader())))
- .expect("response must build succeed");
+ let body = IncomingBody::new(resp, content_length);
+
+ let resp = hr.body(body).expect("response must build succeed");
Ok(resp)
}
diff --git a/core/src/raw/http_util/dns.rs b/core/src/raw/http_util/dns.rs
deleted file mode 100644
index 79a35815..00000000
--- a/core/src/raw/http_util/dns.rs
+++ /dev/null
@@ -1,237 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-use std::collections::HashMap;
-use std::io;
-use std::net::SocketAddr;
-use std::net::ToSocketAddrs;
-use std::sync::Arc;
-use std::time::Duration;
-use std::time::SystemTime;
-
-use futures::future;
-use parking_lot::Mutex;
-
-/// StdDnsResolver uses `getaddrinfo` to query dns.
-#[derive(Default)]
-pub struct StdDnsResolver {
- cache: DnsCache,
-}
-
-impl ureq::Resolver for StdDnsResolver {
- fn resolve(&self, netloc: &str) -> io::Result<Vec<SocketAddr>> {
- if let Some(v) = self.cache.get(netloc) {
- return Ok(v);
- }
-
- ToSocketAddrs::to_socket_addrs(netloc).map(|iter| {
- let res: Vec<_> = iter.collect();
- // Insert into cache if resolved succeeded.
- self.cache.insert(netloc, res.clone());
- res
- })
- }
-}
-
-/// AsyncStdDnsResolver uses `getaddrinfo` to query dns in tokio runtime.
-pub struct AsyncStdDnsResolver {
- cache: DnsCache,
- runtime: Option<tokio::runtime::Runtime>,
-}
-
-impl Default for AsyncStdDnsResolver {
- fn default() -> Self {
- let runtime = {
- let mut builder = tokio::runtime::Builder::new_current_thread();
- builder.enable_all();
-
- builder.build().expect("build dns runtime failed")
- };
-
- Self {
- cache: DnsCache::default(),
- runtime: Some(runtime),
- }
- }
-}
-
-/// Make sure runtime has been dropped in background.
-impl Drop for AsyncStdDnsResolver {
- fn drop(&mut self) {
- let runtime = self.runtime.take().unwrap();
- runtime.shutdown_background();
- }
-}
-
-impl reqwest::dns::Resolve for AsyncStdDnsResolver {
- fn resolve(&self, name: hyper::client::connect::dns::Name) -> reqwest::dns::Resolving {
- if let Some(v) = self.cache.get(name.as_str()) {
- return Box::pin(future::ok(Box::new(v.into_iter()) as reqwest::dns::Addrs));
- }
-
- debug_assert!(self.runtime.is_some(), "runtime must be valid");
- let runtime = self.runtime.as_ref().unwrap().handle().clone();
-
- let cache = self.cache.clone();
- let fut = async move {
- match runtime
- .spawn_blocking(move || {
- // hyper will set port externally, so the port here is
- // just used to construct a correct query.
- //
- // Ref: <https://github.com/hyperium/hyper/blob/4d89adce6122af1650165337d9d814314e7ee409/src/client/connect/http.rs#L322-L359>
- (name.as_str(), 0).to_socket_addrs().map(|iter| {
- let res: Vec<_> = iter.collect();
- // Insert into cache if resolved succeeded.
- cache.insert(name.as_str(), res.clone());
- res
- })
- })
- .await
- {
- Ok(v) => v.map(|v| Box::new(v.into_iter()) as reqwest::dns::Addrs),
- Err(err) => Err(io::Error::new(
- io::ErrorKind::Other,
- format!("spawn dns resolving task failed: {err:?}"),
- )),
- }
- .map_err(|err| Box::new(err) as Box<_>)
- };
-
- Box::pin(fut)
- }
-}
-
-/// TrustDnsResolver will adopt `trust-dns` to query dns in tokio runtime.
-///
-/// We will only use trust dns while we have an async runtime.
-#[cfg(feature = "trust-dns")]
-pub struct AsyncTrustDnsResolver {
- inner: Arc<trust_dns_resolver::TokioAsyncResolver>,
-}
-
-#[cfg(feature = "trust-dns")]
-impl AsyncTrustDnsResolver {
- pub fn new() -> io::Result<Self> {
- let resolver = trust_dns_resolver::TokioAsyncResolver::from_system_conf(
- trust_dns_resolver::TokioHandle,
- )?;
-
- Ok(Self {
- inner: Arc::new(resolver),
- })
- }
-}
-
-#[cfg(feature = "trust-dns")]
-impl reqwest::dns::Resolve for AsyncTrustDnsResolver {
- fn resolve(&self, name: hyper::client::connect::dns::Name) -> reqwest::dns::Resolving {
- let resolver = self.inner.clone();
-
- let fut = async move {
- let lookup = resolver.lookup_ip(name.as_str()).await?;
-
- // hyper will set port externally, so the port here is
- // just used to construct a correct SocketAddr.
- //
- // Ref: <https://github.com/hyperium/hyper/blob/4d89adce6122af1650165337d9d814314e7ee409/src/client/connect/http.rs#L322-L359>
- Ok(Box::new(lookup.into_iter().map(|v| SocketAddr::new(v, 0))) as reqwest::dns::Addrs)
- };
-
- Box::pin(fut)
- }
-}
-
-/// The cache entry we maintained in memory.
-#[derive(Clone)]
-struct DnsCacheEntry {
- value: Vec<SocketAddr>,
- expires_in: SystemTime,
-}
-
-/// DnsCache that used for accessing different storage services.
-///
-/// # Notes
-///
-/// **DON'T USE THIS DNS CACHE OUTSIDE OPENDAL.**
-///
-/// This is not a general dns cache that suitable for all use-cases.
-#[derive(Clone)]
-struct DnsCache {
- inner: Arc<Mutex<HashMap<String, DnsCacheEntry>>>,
-
- /// The most cache entries that we can hold.
- ///
- /// We will use 32 as default value (the same value as trust-dns).
- limits: usize,
- /// Ideally, we need to set the expire based on resolved dns record. But
- /// rust std doesn't give us this choice, so we have to use default one.
- ///
- /// We will use 3600s as the default expire time. In the future, we will
- /// allow user to configure it.
- default_expire: Duration,
-}
-
-impl Default for DnsCache {
- fn default() -> Self {
- DnsCache {
- inner: Arc::default(),
- limits: 32,
- default_expire: Duration::from_secs(3600),
- }
- }
-}
-
-impl DnsCache {
- fn get(&self, domain: &str) -> Option<Vec<SocketAddr>> {
- let mut guard = self.inner.lock();
- match guard.get(domain) {
- None => None,
- Some(entry) => {
- let now = SystemTime::now();
- if entry.expires_in >= now {
- Some(entry.value.clone())
- } else {
- // Remove already expires entry.
- guard.remove(domain);
- None
- }
- }
- }
- }
-
- fn insert(&self, domain: &str, value: Vec<SocketAddr>) {
- let mut guard = self.inner.lock();
-
- // If we are reaching the limits of dns cache entry. We will clean the
- // entire cache to make more space.
- //
- // As described in DnsCache's doc, this limit should never be reached.
- // We expect there only few entries held in cache.
- if guard.len() >= self.limits {
- guard.clear()
- }
-
- guard.insert(
- domain.to_string(),
- DnsCacheEntry {
- value,
- expires_in: SystemTime::now() + self.default_expire,
- },
- );
- }
-}
diff --git a/core/src/raw/http_util/mod.rs b/core/src/raw/http_util/mod.rs
index 54f90fc7..b2878bd0 100644
--- a/core/src/raw/http_util/mod.rs
+++ b/core/src/raw/http_util/mod.rs
@@ -58,5 +58,3 @@ pub use bytes_range::BytesRange;
mod bytes_content_range;
pub use bytes_content_range::BytesContentRange;
-
-mod dns;
diff --git a/core/src/services/s3/backend.rs b/core/src/services/s3/backend.rs
index e1f9d7b8..da00b864 100644
--- a/core/src/services/s3/backend.rs
+++ b/core/src/services/s3/backend.rs
@@ -35,8 +35,6 @@ use http::Request;
use http::Response;
use http::StatusCode;
use log::debug;
-use log::error;
-use log::warn;
use md5::Digest;
use md5::Md5;
use once_cell::sync::Lazy;
@@ -75,7 +73,6 @@ mod constants {
"x-amz-server-side-encryption-customer-key-md5";
pub const X_AMZ_SERVER_SIDE_ENCRYPTION_AWS_KMS_KEY_ID: &str =
"x-amz-server-side-encryption-aws-kms-key-id";
- pub const X_AMZ_BUCKET_REGION: &str = "x-amz-bucket-region";
pub const RESPONSE_CONTENT_DISPOSITION: &str = "response-content-disposition";
pub const RESPONSE_CACHE_CONTROL: &str = "response-cache-control";
@@ -415,9 +412,7 @@ impl S3Builder {
/// Region represent the signing region of this endpoint.
///
/// - If region is set, we will take user's input first.
- /// - If not, We will try to detect region via [RFC-0057: Auto Region](https://github.com/apache/incubator-opendal/blob/main/docs/rfcs/0057-auto-region.md).
- ///
- /// Most of time, region is not need to be set, especially for AWS S3 and minio.
+ /// - If not, we will use `us-east-1` as default.
pub fn region(&mut self, region: &str) -> &mut Self {
if !region.is_empty() {
self.region = Some(region.to_string())
@@ -672,101 +667,6 @@ impl S3Builder {
self
}
- /// Read RFC-0057: Auto Region for detailed behavior.
- ///
- /// - If region is already known, the region will be returned directly.
- /// - If region is not known, we will send API to `{endpoint}/{bucket}` to get
- /// `x-amz-bucket-region`, use `us-east-1` if not found.
- ///
- /// Returning endpoint will trim bucket name:
- ///
- /// - `https://bucket_name.s3.amazonaws.com` => `https://s3.amazonaws.com`
- fn detect_region(&self, client: &HttpClient) -> Result<String> {
- debug_assert!(
- self.region.is_none(),
- "calling detect region with region already know is buggy"
- );
-
- // Builder's bucket must be valid.
- let bucket = self.bucket.as_str();
-
- let mut endpoint = match &self.endpoint {
- Some(endpoint) => {
- if endpoint.starts_with("http") {
- endpoint.to_string()
- } else {
- // Prefix https if endpoint doesn't start with scheme.
- format!("https://{endpoint}")
- }
- }
- None => "https://s3.amazonaws.com".to_string(),
- };
-
- // If endpoint contains bucket name, we should trim them.
- endpoint = endpoint.replace(&format!("//{bucket}."), "//");
-
- let url = format!("{endpoint}/{bucket}");
- debug!("backend detect region with url: {url}");
-
- let req = http::Request::head(&url).body(Body::Empty).map_err(|e| {
- error!("backend detect_region {}: {:?}", url, e);
- Error::new(ErrorKind::Unexpected, "build request for head")
- .with_context("service", Scheme::S3)
- .with_context("url", &url)
- .set_source(e)
- })?;
-
- let res = client.send(req)?;
- let (res, body) = res.into_parts();
-
- // Make sure the body has been consumed so that we can reuse
- // the connection later.
- let _ = body.consume();
-
- debug!(
- "auto detect region got response: status {:?}, header: {:?}",
- res.status, res.headers
- );
- match res.status {
- // The endpoint works, return with not changed endpoint and
- // default region.
- StatusCode::OK | StatusCode::FORBIDDEN => {
- let region = res
- .headers
- .get(constants::X_AMZ_BUCKET_REGION)
- .unwrap_or(&HeaderValue::from_static("us-east-1"))
- .to_str()
- .map_err(|e| {
- Error::new(ErrorKind::Unexpected, "header value is not valid utf-8")
- .set_source(e)
- })?
- .to_string();
- Ok(region)
- }
- // The endpoint should move, return with constructed endpoint
- StatusCode::MOVED_PERMANENTLY => {
- let region = res
- .headers
- .get(constants::X_AMZ_BUCKET_REGION)
- .ok_or_else(|| Error::new(ErrorKind::Unexpected, "region is empty"))?
- .to_str()
- .map_err(|e| {
- Error::new(ErrorKind::Unexpected, "header value is not valid utf-8")
- .set_source(e)
- })?
- .to_string();
-
- Ok(region)
- }
- // Unexpected status code, fallback to default region "us-east-1"
- code => Err(Error::new(
- ErrorKind::Unexpected,
- "can't detect region automatically, unexpected status code got",
- )
- .with_context("status", code.as_str())),
- }
- }
-
/// Build endpoint with given region.
fn build_endpoint(&self, region: &str) -> String {
let bucket = {
@@ -966,17 +866,8 @@ impl Builder for S3Builder {
// region is required to make signer work.
//
// If we don't know region after loading from builder and env.
- // We should try to detect them.
- let region = self
- .detect_region(&client)
- // If we met error during detect region, use "us-east-1"
- // as default instead of returning error.
- .unwrap_or_else(|err| {
- warn!(
- "backend detect region failed for {err:?}, using default region instead"
- );
- "us-east-1".to_string()
- });
+ // We will use `us-east-1` as default.
+ let region = "us-east-1".to_string();
cfg.set_region(®ion);
region
@@ -995,7 +886,6 @@ impl Builder for S3Builder {
signer_builder.credential_loader({
let mut cred_loader = AwsCredentialLoader::new(cfg);
cred_loader = cred_loader.with_allow_anonymous();
- cred_loader = cred_loader.with_client(client.sync_client());
if self.disable_config_load {
// If load config has been disable, we should also disable
// ec2 metadata to avoid leaking permits.
@@ -1764,41 +1654,11 @@ struct DeleteObjectsResultError {
#[cfg(test)]
mod tests {
- use backon::BlockingRetryable;
- use backon::ExponentialBuilder;
use bytes::Buf;
use bytes::Bytes;
use super::*;
- #[test]
- fn test_region() {
- let _ = env_logger::try_init();
-
- let client = HttpClient::new().unwrap();
-
- let endpoint_cases = vec![
- Some("s3.amazonaws.com"),
- Some("https://s3.amazonaws.com"),
- Some("https://s3.us-east-2.amazonaws.com"),
- None,
- ];
-
- for endpoint in endpoint_cases {
- let mut b = S3Builder::default();
- b.bucket("test");
- if let Some(endpoint) = endpoint {
- b.endpoint(endpoint);
- }
-
- let region = { || b.detect_region(&client) }
- .retry(&ExponentialBuilder::default())
- .call()
- .expect("detect region must success");
- assert_eq!(region, "us-east-2");
- }
- }
-
#[test]
fn test_build_endpoint() {
let _ = env_logger::try_init();