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 15:29:27 UTC
[incubator-opendal] 01/01: refactor: Use reqwest blocking client instead of ureq
This is an automated email from the ASF dual-hosted git repository.
xuanwo pushed a commit to branch integrate-with-new-reqsign
in repository https://gitbox.apache.org/repos/asf/incubator-opendal.git
commit 1480e823dff2e99585eebdc796620c818f41f327
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Tue Apr 4 23:29:02 2023 +0800
refactor: Use reqwest blocking client instead of ureq
Signed-off-by: Xuanwo <gi...@xuanwo.io>
---
Cargo.lock | 17 +--
core/Cargo.toml | 8 +-
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, 153 insertions(+), 549 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index eb62479b..43ac86a3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -204,7 +204,7 @@ dependencies = [
"futures-io",
"rustls 0.19.1",
"webpki 0.21.4",
- "webpki-roots 0.21.1",
+ "webpki-roots",
]
[[package]]
@@ -2266,7 +2266,6 @@ dependencies = [
"tracing-opentelemetry",
"tracing-subscriber",
"trust-dns-resolver",
- "ureq",
"uuid",
"wiremock",
]
@@ -4208,13 +4207,8 @@ checksum = "338b31dd1314f68f3aabf3ed57ab922df95ffcd902476ca7ba3c4ce7b908c46d"
dependencies = [
"base64 0.13.1",
"log",
- "native-tls",
"once_cell",
- "rustls 0.20.8",
- "rustls-native-certs",
"url",
- "webpki 0.22.0",
- "webpki-roots 0.22.6",
]
[[package]]
@@ -4444,15 +4438,6 @@ dependencies = [
"webpki 0.21.4",
]
-[[package]]
-name = "webpki-roots"
-version = "0.22.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b6c71e40d7d2c34a5106301fb632274ca37242cd0c9d3e64dbece371a40a2d87"
-dependencies = [
- "webpki 0.22.0",
-]
-
[[package]]
name = "widestring"
version = "0.5.1"
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 4b57fbaa..f43ce0b0 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -45,11 +45,11 @@ docs = []
trust-dns = ["reqwest/trust-dns", "dep:trust-dns-resolver"]
# Enable rustls for TLS support
-rustls = ["reqwest/rustls-tls-native-roots", "ureq/tls", "ureq/native-certs"]
+rustls = ["reqwest/rustls-tls-native-roots"]
# Enable native-tls for TLS support
-native-tls = ["reqwest/native-tls", "ureq/native-tls"]
+native-tls = ["reqwest/native-tls"]
# Enable vendored native-tls for TLS support
-native-tls-vendored = ["reqwest/native-tls-vendored", "ureq/native-tls"]
+native-tls-vendored = ["reqwest/native-tls-vendored"]
# Enable all layers.
layers-all = ["layers-chaos", "layers-metrics", "layers-tracing"]
@@ -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"] }
@@ -134,7 +135,6 @@ time = { version = "0.3", features = ["serde"] }
tokio = { version = "1.27", features = ["fs"] }
tracing = { version = "0.1", optional = true }
trust-dns-resolver = { version = "0.22", optional = true }
-ureq = { version = "2", default-features = false }
uuid = { version = "1", features = ["serde", "v4"] }
[dev-dependencies]
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();