You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "tustvold (via GitHub)" <gi...@apache.org> on 2025/02/23 22:04:40 UTC
[PR] Http client abstraction [arrow-rs]
tustvold opened a new pull request, #7183:
URL: https://github.com/apache/arrow-rs/pull/7183
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #6056
Related to #6818
Related to #7171
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
See tickets
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!---
If there are any breaking changes to public APIs, please call them out.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Decouple ObjectStore from Reqwest [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7183:
URL: https://github.com/apache/arrow-rs/pull/7183#discussion_r1967363421
##########
object_store/src/client/connection.rs:
##########
@@ -0,0 +1,222 @@
+// 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 crate::client::body::{HttpRequest, HttpResponse};
+use crate::client::builder::{HttpRequestBuilder, RequestBuilderError};
+use crate::client::HttpResponseBody;
+use crate::ClientOptions;
+use async_trait::async_trait;
+use http::{Method, Uri};
+use http_body_util::BodyExt;
+use std::error::Error;
+use std::sync::Arc;
+
+/// An HTTP protocol error
+///
+/// Clients should return this when an HTTP request fails to be completed, e.g. because
+/// of a connection issue. This does **not** include HTTP requests that are return
Review Comment:
FWIW the way reqwest conflates these two scenarios I have always found immensely confusing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Decouple ObjectStore from Reqwest [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7183:
URL: https://github.com/apache/arrow-rs/pull/7183#discussion_r1967366010
##########
object_store/src/client/retry.rs:
##########
@@ -284,164 +334,97 @@ impl RetryableRequest {
return Ok(r);
}
- let status = r.status();
- let headers = r.headers().clone();
-
- let bytes = r.bytes().await.map_err(|e| Error::Reqwest {
- retries,
- max_retries,
- elapsed: now.elapsed(),
- retry_timeout,
- source: e,
- })?;
-
- let response_body = String::from_utf8_lossy(&bytes);
- debug!("Checking for error in response_body: {}", response_body);
+ let (parts, body) = r.into_parts();
+ let body = match body.text().await {
+ Ok(body) => body,
+ Err(e) => return Err(ctx.err(RequestError::Http(e))),
+ };
- if !body_contains_error(&response_body) {
+ if !body_contains_error(&body) {
// Success response and no error, clone and return response
- let mut success_response = hyper::Response::new(bytes);
- *success_response.status_mut() = status;
- *success_response.headers_mut() = headers;
-
- return Ok(reqwest::Response::from(success_response));
+ return Ok(HttpResponse::from_parts(parts, body.into()));
} else {
// Retry as if this was a 5xx response
- if retries == max_retries || now.elapsed() > retry_timeout {
- return Err(Error::Server {
- body: Some(response_body.into_owned()),
- status,
- });
+ if ctx.exhausted() {
+ return Err(ctx.err(RequestError::Response { body, status }));
}
let sleep = backoff.next();
- retries += 1;
+ ctx.retries += 1;
info!(
"Encountered a response status of {} but body contains Error, backing off for {} seconds, retry {} of {}",
status,
sleep.as_secs_f32(),
- retries,
- max_retries,
+ ctx.retries,
+ ctx.max_retries,
);
tokio::time::sleep(sleep).await;
}
- }
- Ok(r) if r.status() == StatusCode::NOT_MODIFIED => {
- return Err(Error::Client {
- body: None,
- status: StatusCode::NOT_MODIFIED,
- })
- }
- Ok(r) => {
- let is_bare_redirect =
- r.status().is_redirection() && !r.headers().contains_key(LOCATION);
+ } else if status == StatusCode::NOT_MODIFIED {
+ return Err(ctx.err(RequestError::Status { status, body: None }));
+ } else if status.is_redirection() {
+ let is_bare_redirect = !r.headers().contains_key(LOCATION);
return match is_bare_redirect {
- true => Err(Error::BareRedirect),
- // Not actually sure if this is reachable, but here for completeness
- false => Err(Error::Client {
+ true => Err(ctx.err(RequestError::BareRedirect)),
+ false => Err(ctx.err(RequestError::Status {
body: None,
status: r.status(),
- }),
+ })),
};
- }
- Err(e) => {
- let e = sanitize_err(e);
+ } else {
let status = r.status();
- if retries == max_retries
- || now.elapsed() > retry_timeout
+ if ctx.exhausted()
|| !(status.is_server_error()
|| (self.retry_on_conflict && status == StatusCode::CONFLICT))
{
- return Err(match status.is_client_error() {
- true => match r.text().await {
- Ok(body) => Error::Client {
- body: Some(body).filter(|b| !b.is_empty()),
+ let source = match status.is_client_error() {
+ true => match r.into_body().text().await {
+ Ok(body) => RequestError::Status {
status,
+ body: Some(body),
},
- Err(e) => Error::Reqwest {
- retries,
- max_retries,
- elapsed: now.elapsed(),
- retry_timeout,
- source: e,
- },
+ Err(e) => RequestError::Http(e),
},
- false => Error::Reqwest {
- retries,
- max_retries,
- elapsed: now.elapsed(),
- retry_timeout,
- source: e,
- },
- });
- }
+ false => RequestError::Status { status, body: None },
+ };
+ return Err(ctx.err(source));
+ };
let sleep = backoff.next();
- retries += 1;
+ ctx.retries += 1;
info!(
- "Encountered server error, backing off for {} seconds, retry {} of {}: {}",
+ "Encountered server error, backing off for {} seconds, retry {} of {}",
sleep.as_secs_f32(),
- retries,
- max_retries,
- e,
+ ctx.retries,
+ ctx.max_retries,
);
tokio::time::sleep(sleep).await;
}
- },
+ }
Err(e) => {
- let e = sanitize_err(e);
-
- let mut do_retry = false;
- if e.is_connect()
- || e.is_body()
- || (e.is_request() && !e.is_timeout())
- || (is_idempotent && e.is_timeout())
+ // let e = sanitize_err(e);
+
+ let do_retry = match e.kind() {
+ HttpErrorKind::Connect | HttpErrorKind::Request => true, // Request not sent, can retry
+ HttpErrorKind::Timeout => is_idempotent,
+ HttpErrorKind::Unknown
+ | HttpErrorKind::Interrupted
+ | HttpErrorKind::Decode => false,
+ };
+
+ if ctx.retries == ctx.max_retries
+ || ctx.start.elapsed() > ctx.retry_timeout
+ || !do_retry
{
- do_retry = true
- } else {
- let mut source = e.source();
- while let Some(e) = source {
- if let Some(e) = e.downcast_ref::<hyper::Error>() {
- do_retry = e.is_closed()
- || e.is_incomplete_message()
- || e.is_body_write_aborted()
- || (is_idempotent && e.is_timeout());
- break;
- }
- if let Some(e) = e.downcast_ref::<std::io::Error>() {
- if e.kind() == std::io::ErrorKind::TimedOut {
- do_retry = is_idempotent;
- } else {
- do_retry = matches!(
- e.kind(),
- std::io::ErrorKind::ConnectionReset
- | std::io::ErrorKind::ConnectionAborted
- | std::io::ErrorKind::BrokenPipe
- | std::io::ErrorKind::UnexpectedEof
Review Comment:
FWIW we no longer retry these variants, as tbh I don't think it is correct to do so
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Decouple ObjectStore from Reqwest [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7183:
URL: https://github.com/apache/arrow-rs/pull/7183#discussion_r1966911968
##########
object_store/src/client/body.rs:
##########
@@ -0,0 +1,213 @@
+// 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 crate::client::connection::{HttpError, HttpErrorKind};
+use crate::{collect_bytes, PutPayload};
+use bytes::Bytes;
+use futures::stream::BoxStream;
+use futures::StreamExt;
+use http_body_util::combinators::BoxBody;
+use http_body_util::{BodyExt, Full};
+use hyper::body::{Body, Frame, SizeHint};
+use serde::de::DeserializeOwned;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// An HTTP Request
+pub type HttpRequest = http::Request<HttpRequestBody>;
+
+/// The [`Body`] of an [`HttpRequest`]
+#[derive(Debug, Clone)]
+pub struct HttpRequestBody(Inner);
Review Comment:
We define fixed body types as we need HttpClient to be dyn-compatible (it also makes things slightly easier to follow)
##########
object_store/src/client/connection.rs:
##########
@@ -0,0 +1,222 @@
+// 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 crate::client::body::{HttpRequest, HttpResponse};
+use crate::client::builder::{HttpRequestBuilder, RequestBuilderError};
+use crate::client::HttpResponseBody;
+use crate::ClientOptions;
+use async_trait::async_trait;
+use http::{Method, Uri};
+use http_body_util::BodyExt;
+use std::error::Error;
+use std::sync::Arc;
+
+/// An HTTP protocol error
+///
+/// Clients should return this when an HTTP request fails to be completed, e.g. because
+/// of a connection issue. This does **not** include HTTP requests that are return
+/// non 2xx Status Codes, as these should instead be returned as an [`HttpResponse`]
+/// with the appropriate status code set.
+#[derive(Debug, thiserror::Error)]
+#[error("HTTP error: {source}")]
+pub struct HttpError {
+ kind: HttpErrorKind,
+ #[source]
+ source: Box<dyn Error + Send + Sync>,
+}
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum HttpErrorKind {
+ /// Request timed out
+ Timeout,
+ /// An error occurred whilst connecting to the remote
+ Connect,
+ /// An error occurred whilst making the request
+ Request,
+ /// An error occurred whilst decoding the response
+ Decode,
+ /// The request was aborted
+ Interrupted,
+ /// An unknown error occurred
+ Unknown,
+}
+
+impl HttpError {
+ /// Create a new [`HttpError`] with the optional status code
+ pub fn new<E>(kind: HttpErrorKind, e: E) -> Self
+ where
+ E: Error + Send + Sync + 'static,
+ {
+ Self {
+ kind,
+ source: Box::new(e),
+ }
+ }
+
+ pub(crate) fn reqwest(e: reqwest::Error) -> Self {
+ let mut kind = if e.is_timeout() {
+ HttpErrorKind::Timeout
+ } else if e.is_connect() {
+ HttpErrorKind::Connect
+ } else if e.is_decode() {
+ HttpErrorKind::Decode
+ } else {
+ HttpErrorKind::Unknown
+ };
+
+ // Reqwest error variants aren't great, attempt to refine them
+ let mut source = e.source();
Review Comment:
This logic is moved out of retry.rs and reframed in terms of HttpErrorKind
##########
object_store/src/client/builder.rs:
##########
@@ -0,0 +1,275 @@
+// 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 crate::client::connection::HttpErrorKind;
+use crate::client::{HttpClient, HttpError, HttpRequest, HttpRequestBody};
+use http::header::{Entry, InvalidHeaderName, InvalidHeaderValue, OccupiedEntry};
+use http::uri::{InvalidUri, PathAndQuery};
+use http::{HeaderMap, HeaderName, HeaderValue, Method, Uri};
+use serde::Serialize;
+use std::borrow::Borrow;
+
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum RequestBuilderError {
+ #[error("Invalid URI")]
+ InvalidUri(#[from] InvalidUri),
+
+ #[error("Invalid Header Value")]
+ InvalidHeaderValue(#[from] InvalidHeaderValue),
+
+ #[error("Invalid Header Name")]
+ InvalidHeaderName(#[from] InvalidHeaderName),
+
+ #[error("JSON serialization error")]
+ SerdeJson(#[from] serde_json::Error),
+
+ #[error("URL serialization error")]
+ SerdeUrl(#[from] serde_urlencoded::ser::Error),
+}
+
+impl From<RequestBuilderError> for HttpError {
+ fn from(value: RequestBuilderError) -> Self {
+ Self::new(HttpErrorKind::Request, value)
+ }
+}
+
+impl From<std::convert::Infallible> for RequestBuilderError {
+ fn from(value: std::convert::Infallible) -> Self {
+ match value {}
+ }
+}
+
+pub(crate) struct HttpRequestBuilder {
Review Comment:
This is a crate-private builder that provides an interface similar to `reqwest` but which can then be used with the `HttpClient`. Unfortunately `reqwest::Request` is largely opaque and I couldn't see an obvious way to make use of it.
##########
object_store/src/client/connection.rs:
##########
@@ -0,0 +1,222 @@
+// 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 crate::client::body::{HttpRequest, HttpResponse};
+use crate::client::builder::{HttpRequestBuilder, RequestBuilderError};
+use crate::client::HttpResponseBody;
+use crate::ClientOptions;
+use async_trait::async_trait;
+use http::{Method, Uri};
+use http_body_util::BodyExt;
+use std::error::Error;
+use std::sync::Arc;
+
+/// An HTTP protocol error
+///
+/// Clients should return this when an HTTP request fails to be completed, e.g. because
+/// of a connection issue. This does **not** include HTTP requests that are return
+/// non 2xx Status Codes, as these should instead be returned as an [`HttpResponse`]
+/// with the appropriate status code set.
+#[derive(Debug, thiserror::Error)]
+#[error("HTTP error: {source}")]
+pub struct HttpError {
+ kind: HttpErrorKind,
+ #[source]
+ source: Box<dyn Error + Send + Sync>,
+}
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum HttpErrorKind {
Review Comment:
This allows the retry logic to work on top of arbitrary HttpClients
##########
object_store/src/client/builder.rs:
##########
@@ -0,0 +1,275 @@
+// 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 crate::client::connection::HttpErrorKind;
+use crate::client::{HttpClient, HttpError, HttpRequest, HttpRequestBody};
+use http::header::{Entry, InvalidHeaderName, InvalidHeaderValue, OccupiedEntry};
+use http::uri::{InvalidUri, PathAndQuery};
+use http::{HeaderMap, HeaderName, HeaderValue, Method, Uri};
+use serde::Serialize;
+use std::borrow::Borrow;
+
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum RequestBuilderError {
+ #[error("Invalid URI")]
+ InvalidUri(#[from] InvalidUri),
+
+ #[error("Invalid Header Value")]
+ InvalidHeaderValue(#[from] InvalidHeaderValue),
+
+ #[error("Invalid Header Name")]
+ InvalidHeaderName(#[from] InvalidHeaderName),
+
+ #[error("JSON serialization error")]
+ SerdeJson(#[from] serde_json::Error),
+
+ #[error("URL serialization error")]
+ SerdeUrl(#[from] serde_urlencoded::ser::Error),
+}
+
+impl From<RequestBuilderError> for HttpError {
+ fn from(value: RequestBuilderError) -> Self {
+ Self::new(HttpErrorKind::Request, value)
+ }
+}
+
+impl From<std::convert::Infallible> for RequestBuilderError {
+ fn from(value: std::convert::Infallible) -> Self {
+ match value {}
+ }
+}
+
+pub(crate) struct HttpRequestBuilder {
+ client: HttpClient,
+ request: Result<HttpRequest, RequestBuilderError>,
+}
+
+impl HttpRequestBuilder {
+ pub(crate) fn new(client: HttpClient) -> Self {
+ Self {
+ client,
+ request: Ok(HttpRequest::new(HttpRequestBody::empty())),
+ }
+ }
+
+ pub(crate) fn from_parts(client: HttpClient, request: HttpRequest) -> Self {
+ Self {
+ client,
+ request: Ok(request),
+ }
+ }
+
+ pub(crate) fn method(mut self, method: Method) -> Self {
+ if let Ok(r) = &mut self.request {
+ *r.method_mut() = method;
+ }
+ self
+ }
+
+ pub(crate) fn uri<U>(mut self, url: U) -> Self
+ where
+ U: TryInto<Uri>,
+ U::Error: Into<RequestBuilderError>,
+ {
+ match (url.try_into(), &mut self.request) {
+ (Ok(uri), Ok(r)) => *r.uri_mut() = uri,
+ (Err(e), Ok(_)) => self.request = Err(e.into()),
+ (_, Err(_)) => {}
+ }
+ self
+ }
+
+ pub(crate) fn header<K, V>(mut self, name: K, value: V) -> Self
+ where
+ K: TryInto<HeaderName>,
+ K::Error: Into<RequestBuilderError>,
+ V: TryInto<HeaderValue>,
+ V::Error: Into<RequestBuilderError>,
+ {
+ match (name.try_into(), value.try_into(), &mut self.request) {
+ (Ok(name), Ok(value), Ok(r)) => {
+ r.headers_mut().insert(name, value);
+ }
+ (Err(e), _, Ok(_)) => self.request = Err(e.into()),
+ (_, Err(e), Ok(_)) => self.request = Err(e.into()),
+ (_, _, Err(_)) => {}
+ }
+ self
+ }
+
+ pub(crate) fn headers(mut self, headers: HeaderMap) -> Self {
+ if let Ok(ref mut req) = self.request {
+ // IntoIter of HeaderMap yields (Option<HeaderName>, HeaderValue).
+ // The first time a name is yielded, it will be Some(name), and if
+ // there are more values with the same name, the next yield will be
+ // None.
+
+ let mut prev_entry: Option<OccupiedEntry<'_, _>> = None;
+ for (key, value) in headers {
+ match key {
+ Some(key) => match req.headers_mut().entry(key) {
+ Entry::Occupied(mut e) => {
+ e.insert(value);
+ prev_entry = Some(e);
+ }
+ Entry::Vacant(e) => {
+ let e = e.insert_entry(value);
+ prev_entry = Some(e);
+ }
+ },
+ None => match prev_entry {
+ Some(ref mut entry) => {
+ entry.append(value);
+ }
+ None => unreachable!("HeaderMap::into_iter yielded None first"),
+ },
+ }
+ }
+ }
+ self
+ }
+
+ #[cfg(feature = "gcp")]
+ pub(crate) fn bearer_auth(mut self, token: &str) -> Self {
+ let value = HeaderValue::try_from(format!("Bearer {}", token));
+ match (value, &mut self.request) {
+ (Ok(mut v), Ok(r)) => {
+ v.set_sensitive(true);
+ r.headers_mut().insert(http::header::AUTHORIZATION, v);
+ }
+ (Err(e), Ok(_)) => self.request = Err(e.into()),
+ (_, Err(_)) => {}
+ }
+ self
+ }
+
+ #[cfg(any(feature = "aws", feature = "gcp"))]
+ pub(crate) fn json<S: Serialize>(mut self, s: S) -> Self {
+ match (serde_json::to_vec(&s), &mut self.request) {
+ (Ok(json), Ok(request)) => {
+ *request.body_mut() = json.into();
+ }
+ (Err(e), Ok(_)) => self.request = Err(e.into()),
+ (_, Err(_)) => {}
+ }
+ self
+ }
+
+ #[cfg(any(feature = "aws", feature = "gcp", feature = "azure"))]
+ pub(crate) fn query<T: Serialize + ?Sized>(mut self, query: &T) -> Self {
+ let mut error = None;
+ if let Ok(ref mut req) = self.request {
+ let mut out = format!("{}?", req.uri().path());
+ let mut encoder = form_urlencoded::Serializer::new(&mut out);
+ let serializer = serde_urlencoded::Serializer::new(&mut encoder);
+
+ if let Err(err) = query.serialize(serializer) {
+ error = Some(err.into());
+ }
+
+ match PathAndQuery::from_maybe_shared(out) {
+ Ok(p) => {
+ let mut parts = req.uri().clone().into_parts();
+ parts.path_and_query = Some(p);
+ *req.uri_mut() = Uri::from_parts(parts).unwrap();
+ }
+ Err(err) => error = Some(err.into()),
+ }
+ }
+ if let Some(err) = error {
+ self.request = Err(err);
+ }
+ self
+ }
+
+ #[cfg(any(feature = "gcp", feature = "azure"))]
+ pub(crate) fn form<T: Serialize>(mut self, form: T) -> Self {
+ let mut error = None;
+ if let Ok(ref mut req) = self.request {
+ match serde_urlencoded::to_string(form) {
+ Ok(body) => {
+ req.headers_mut().insert(
+ http::header::CONTENT_TYPE,
+ HeaderValue::from_static("application/x-www-form-urlencoded"),
+ );
+ *req.body_mut() = body.into();
+ }
+ Err(err) => error = Some(err.into()),
+ }
+ }
+ if let Some(err) = error {
+ self.request = Err(err);
+ }
+ self
+ }
+
+ pub(crate) fn body(mut self, b: impl Into<HttpRequestBody>) -> Self {
+ if let Ok(r) = &mut self.request {
+ *r.body_mut() = b.into();
+ }
+ self
+ }
+
+ pub(crate) fn into_parts(self) -> (HttpClient, Result<HttpRequest, RequestBuilderError>) {
+ (self.client, self.request)
+ }
+}
+
+pub(crate) fn add_query_pairs<I, K, V>(uri: &mut Uri, query_pairs: I)
Review Comment:
It is a shame that http::Request uses http::Uri which lacks the ergonomic niceities of Url, so we require some additional shenanigans
##########
object_store/src/client/retry.rs:
##########
@@ -460,12 +443,12 @@ pub(crate) trait RetryExt {
/// # Panic
///
/// This will panic if the request body is a stream
- fn send_retry(self, config: &RetryConfig) -> BoxFuture<'static, Result<Response>>;
+ fn send_retry(self, config: &RetryConfig) -> BoxFuture<'static, Result<HttpResponse>>;
}
-impl RetryExt for reqwest::RequestBuilder {
+impl RetryExt for HttpRequestBuilder {
Review Comment:
We could theoretically remove the extension trait, as HttpRequestBuilder, is not a foreign type. I opted to keep it for now though.
##########
object_store/src/client/retry.rs:
##########
@@ -18,60 +18,118 @@
//! A shared HTTP client implementation incorporating retries
use crate::client::backoff::{Backoff, BackoffConfig};
+use crate::client::builder::HttpRequestBuilder;
+use crate::client::connection::HttpErrorKind;
+use crate::client::{HttpClient, HttpError, HttpRequest, HttpResponse};
use crate::PutPayload;
use futures::future::BoxFuture;
+use http::{Method, Uri};
use reqwest::header::LOCATION;
-use reqwest::{Client, Request, Response, StatusCode};
-use std::error::Error as StdError;
+use reqwest::StatusCode;
use std::time::{Duration, Instant};
-use tracing::{debug, info};
+use tracing::info;
/// Retry request error
#[derive(Debug, thiserror::Error)]
-pub enum Error {
- #[error("Received redirect without LOCATION, this normally indicates an incorrectly configured region")]
+pub struct RetryError {
+ method: Method,
+ uri: Option<Uri>,
+ retries: usize,
+ max_retries: usize,
+ elapsed: Duration,
+ retry_timeout: Duration,
+ inner: RequestError,
+}
+
+impl std::fmt::Display for RetryError {
Review Comment:
This somewhat lays the ground work for #6377
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Decouple ObjectStore from Reqwest [arrow-rs]
Posted by "crepererum (via GitHub)" <gi...@apache.org>.
crepererum commented on PR #7183:
URL: https://github.com/apache/arrow-rs/pull/7183#issuecomment-2677948064
Is it right to say that instead of `reqwest`, we now settle on
- `hyper`
- `http`
- custom Error type
- custom service type
? I wonder if we need the custom service type though or if we could use something like `tower` -- which `reqwest` also uses under the hood.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Decouple ObjectStore from Reqwest [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7183:
URL: https://github.com/apache/arrow-rs/pull/7183#discussion_r1967366010
##########
object_store/src/client/retry.rs:
##########
@@ -284,164 +334,97 @@ impl RetryableRequest {
return Ok(r);
}
- let status = r.status();
- let headers = r.headers().clone();
-
- let bytes = r.bytes().await.map_err(|e| Error::Reqwest {
- retries,
- max_retries,
- elapsed: now.elapsed(),
- retry_timeout,
- source: e,
- })?;
-
- let response_body = String::from_utf8_lossy(&bytes);
- debug!("Checking for error in response_body: {}", response_body);
+ let (parts, body) = r.into_parts();
+ let body = match body.text().await {
+ Ok(body) => body,
+ Err(e) => return Err(ctx.err(RequestError::Http(e))),
+ };
- if !body_contains_error(&response_body) {
+ if !body_contains_error(&body) {
// Success response and no error, clone and return response
- let mut success_response = hyper::Response::new(bytes);
- *success_response.status_mut() = status;
- *success_response.headers_mut() = headers;
-
- return Ok(reqwest::Response::from(success_response));
+ return Ok(HttpResponse::from_parts(parts, body.into()));
} else {
// Retry as if this was a 5xx response
- if retries == max_retries || now.elapsed() > retry_timeout {
- return Err(Error::Server {
- body: Some(response_body.into_owned()),
- status,
- });
+ if ctx.exhausted() {
+ return Err(ctx.err(RequestError::Response { body, status }));
}
let sleep = backoff.next();
- retries += 1;
+ ctx.retries += 1;
info!(
"Encountered a response status of {} but body contains Error, backing off for {} seconds, retry {} of {}",
status,
sleep.as_secs_f32(),
- retries,
- max_retries,
+ ctx.retries,
+ ctx.max_retries,
);
tokio::time::sleep(sleep).await;
}
- }
- Ok(r) if r.status() == StatusCode::NOT_MODIFIED => {
- return Err(Error::Client {
- body: None,
- status: StatusCode::NOT_MODIFIED,
- })
- }
- Ok(r) => {
- let is_bare_redirect =
- r.status().is_redirection() && !r.headers().contains_key(LOCATION);
+ } else if status == StatusCode::NOT_MODIFIED {
+ return Err(ctx.err(RequestError::Status { status, body: None }));
+ } else if status.is_redirection() {
+ let is_bare_redirect = !r.headers().contains_key(LOCATION);
return match is_bare_redirect {
- true => Err(Error::BareRedirect),
- // Not actually sure if this is reachable, but here for completeness
- false => Err(Error::Client {
+ true => Err(ctx.err(RequestError::BareRedirect)),
+ false => Err(ctx.err(RequestError::Status {
body: None,
status: r.status(),
- }),
+ })),
};
- }
- Err(e) => {
- let e = sanitize_err(e);
+ } else {
let status = r.status();
- if retries == max_retries
- || now.elapsed() > retry_timeout
+ if ctx.exhausted()
|| !(status.is_server_error()
|| (self.retry_on_conflict && status == StatusCode::CONFLICT))
{
- return Err(match status.is_client_error() {
- true => match r.text().await {
- Ok(body) => Error::Client {
- body: Some(body).filter(|b| !b.is_empty()),
+ let source = match status.is_client_error() {
+ true => match r.into_body().text().await {
+ Ok(body) => RequestError::Status {
status,
+ body: Some(body),
},
- Err(e) => Error::Reqwest {
- retries,
- max_retries,
- elapsed: now.elapsed(),
- retry_timeout,
- source: e,
- },
+ Err(e) => RequestError::Http(e),
},
- false => Error::Reqwest {
- retries,
- max_retries,
- elapsed: now.elapsed(),
- retry_timeout,
- source: e,
- },
- });
- }
+ false => RequestError::Status { status, body: None },
+ };
+ return Err(ctx.err(source));
+ };
let sleep = backoff.next();
- retries += 1;
+ ctx.retries += 1;
info!(
- "Encountered server error, backing off for {} seconds, retry {} of {}: {}",
+ "Encountered server error, backing off for {} seconds, retry {} of {}",
sleep.as_secs_f32(),
- retries,
- max_retries,
- e,
+ ctx.retries,
+ ctx.max_retries,
);
tokio::time::sleep(sleep).await;
}
- },
+ }
Err(e) => {
- let e = sanitize_err(e);
-
- let mut do_retry = false;
- if e.is_connect()
- || e.is_body()
- || (e.is_request() && !e.is_timeout())
- || (is_idempotent && e.is_timeout())
+ // let e = sanitize_err(e);
+
+ let do_retry = match e.kind() {
+ HttpErrorKind::Connect | HttpErrorKind::Request => true, // Request not sent, can retry
+ HttpErrorKind::Timeout => is_idempotent,
+ HttpErrorKind::Unknown
+ | HttpErrorKind::Interrupted
+ | HttpErrorKind::Decode => false,
+ };
+
+ if ctx.retries == ctx.max_retries
+ || ctx.start.elapsed() > ctx.retry_timeout
+ || !do_retry
{
- do_retry = true
- } else {
- let mut source = e.source();
- while let Some(e) = source {
- if let Some(e) = e.downcast_ref::<hyper::Error>() {
- do_retry = e.is_closed()
- || e.is_incomplete_message()
- || e.is_body_write_aborted()
- || (is_idempotent && e.is_timeout());
- break;
- }
- if let Some(e) = e.downcast_ref::<std::io::Error>() {
- if e.kind() == std::io::ErrorKind::TimedOut {
- do_retry = is_idempotent;
- } else {
- do_retry = matches!(
- e.kind(),
- std::io::ErrorKind::ConnectionReset
- | std::io::ErrorKind::ConnectionAborted
- | std::io::ErrorKind::BrokenPipe
- | std::io::ErrorKind::UnexpectedEof
Review Comment:
FWIW we no longer retry these variants, as tbh I don't think it is correct to do so
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Decouple ObjectStore from Reqwest [arrow-rs]
Posted by "crepererum (via GitHub)" <gi...@apache.org>.
crepererum commented on code in PR #7183:
URL: https://github.com/apache/arrow-rs/pull/7183#discussion_r1971549942
##########
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"]
Review Comment:
I think we can do that in a follow-up, but it might be good to settle on the explicit style for all references, i.e. prefixing all dependency crates with `dep:`.
##########
object_store/src/client/connection.rs:
##########
@@ -0,0 +1,237 @@
+// 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 crate::client::body::{HttpRequest, HttpResponse};
+use crate::client::builder::{HttpRequestBuilder, RequestBuilderError};
+use crate::client::HttpResponseBody;
+use crate::ClientOptions;
+use async_trait::async_trait;
+use http::{Method, Uri};
+use http_body_util::BodyExt;
+use std::error::Error;
+use std::sync::Arc;
+
+/// An HTTP protocol error
+///
+/// Clients should return this when an HTTP request fails to be completed, e.g. because
+/// of a connection issue. This does **not** include HTTP requests that are return
+/// non 2xx Status Codes, as these should instead be returned as an [`HttpResponse`]
+/// with the appropriate status code set.
+#[derive(Debug, thiserror::Error)]
+#[error("HTTP error: {source}")]
+pub struct HttpError {
+ kind: HttpErrorKind,
+ #[source]
+ source: Box<dyn Error + Send + Sync>,
+}
+
+/// Identifies the kind of [`HttpError`]
+///
+/// This is used, among other things, to determine if a request can be retried
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum HttpErrorKind {
+ /// An error occurred whilst connecting to the remote
+ ///
+ /// Will be automatically retried
+ Connect,
+ /// An error occurred whilst making the request
+ ///
+ /// Will be automatically retried
+ Request,
+ /// Request timed out
+ ///
+ /// Will be automatically retried if the request is idempotent
+ Timeout,
+ /// The request was aborted
+ ///
+ /// Will be automatically retried if the request is idempotent
+ Interrupted,
+ /// An error occurred whilst decoding the response
+ ///
+ /// Will not be automatically retried
+ Decode,
+ /// An unknown error occurred
+ ///
+ /// Will not be automatically retried
+ Unknown,
+}
+
+impl HttpError {
+ /// Create a new [`HttpError`] with the optional status code
+ pub fn new<E>(kind: HttpErrorKind, e: E) -> Self
+ where
+ E: Error + Send + Sync + 'static,
+ {
+ Self {
+ kind,
+ source: Box::new(e),
+ }
+ }
+
+ pub(crate) fn reqwest(e: reqwest::Error) -> Self {
+ let mut kind = if e.is_timeout() {
+ HttpErrorKind::Timeout
+ } else if e.is_connect() {
+ HttpErrorKind::Connect
+ } else if e.is_decode() {
+ HttpErrorKind::Decode
+ } else {
+ HttpErrorKind::Unknown
+ };
+
+ // Reqwest error variants aren't great, attempt to refine them
+ let mut source = e.source();
+ while let Some(e) = source {
+ if let Some(e) = e.downcast_ref::<hyper::Error>() {
+ if e.is_closed() || e.is_incomplete_message() || e.is_body_write_aborted() {
+ kind = HttpErrorKind::Request;
+ } else if e.is_timeout() {
+ kind = HttpErrorKind::Timeout;
+ }
+ break;
+ }
+ if let Some(e) = e.downcast_ref::<std::io::Error>() {
+ match e.kind() {
+ std::io::ErrorKind::TimedOut => kind = HttpErrorKind::Timeout,
+ std::io::ErrorKind::ConnectionAborted
+ | std::io::ErrorKind::BrokenPipe
+ | std::io::ErrorKind::UnexpectedEof => kind = HttpErrorKind::Interrupted,
+ _ => {}
+ }
+ break;
+ }
+ source = e.source();
+ }
+ Self {
+ kind,
+ // We strip URL as it will be included by RetryError if not sensitive
+ source: Box::new(e.without_url()),
+ }
+ }
+
+ /// Returns the [`HttpErrorKind`]
+ pub fn kind(&self) -> HttpErrorKind {
+ self.kind
+ }
+}
+
+/// An asynchronous function from a [`HttpRequest`] to a [`HttpResponse`].
+#[async_trait]
+pub trait HttpService: std::fmt::Debug + Send + Sync + 'static {
+ /// Perform [`HttpRequest`] returning [`HttpResponse`]
+ async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError>;
+}
+
+/// An HTTP client
+#[derive(Debug, Clone)]
+pub struct HttpClient(Arc<dyn HttpService>);
+
+impl HttpClient {
+ /// Create a new [`HttpClient`] from an [`HttpService`]
+ pub fn new(service: impl HttpService + 'static) -> Self {
+ Self(Arc::new(service))
+ }
+
+ /// Performs [`HttpRequest`] using this client
+ pub async fn execute(&self, request: HttpRequest) -> Result<HttpResponse, HttpError> {
+ self.0.call(request).await
+ }
+
+ #[allow(unused)]
+ pub(crate) fn get<U>(&self, url: U) -> HttpRequestBuilder
+ where
+ U: TryInto<Uri>,
+ U::Error: Into<RequestBuilderError>,
+ {
+ self.request(Method::GET, url)
+ }
+
+ #[allow(unused)]
+ pub(crate) fn post<U>(&self, url: U) -> HttpRequestBuilder
+ where
+ U: TryInto<Uri>,
+ U::Error: Into<RequestBuilderError>,
+ {
+ self.request(Method::POST, url)
+ }
+
+ #[allow(unused)]
+ pub(crate) fn put<U>(&self, url: U) -> HttpRequestBuilder
+ where
+ U: TryInto<Uri>,
+ U::Error: Into<RequestBuilderError>,
+ {
+ self.request(Method::PUT, url)
+ }
+
+ #[allow(unused)]
+ pub(crate) fn delete<U>(&self, url: U) -> HttpRequestBuilder
+ where
+ U: TryInto<Uri>,
+ U::Error: Into<RequestBuilderError>,
+ {
+ self.request(Method::DELETE, url)
+ }
+
+ pub(crate) fn request<U>(&self, method: Method, url: U) -> HttpRequestBuilder
+ where
+ U: TryInto<Uri>,
+ U::Error: Into<RequestBuilderError>,
+ {
+ HttpRequestBuilder::new(self.clone())
+ .uri(url)
+ .method(method)
+ }
+}
+
+#[async_trait]
+impl HttpService for reqwest::Client {
+ async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError> {
+ let (parts, body) = req.into_parts();
+
+ let url = parts.uri.to_string().parse().unwrap();
+ let mut req = reqwest::Request::new(parts.method, url);
+ *req.headers_mut() = parts.headers;
+ *req.body_mut() = Some(body.into_reqwest());
+
+ let r = self.execute(req).await.map_err(HttpError::reqwest)?;
+ let res: http::Response<reqwest::Body> = r.into();
+ let (parts, body) = res.into_parts();
+
+ let body = HttpResponseBody::new(body.map_err(HttpError::reqwest));
+ Ok(HttpResponse::from_parts(parts, body))
+ }
+}
+
+/// A factory for [`HttpClient`]
+pub trait HttpConnector: std::fmt::Debug + Send + Sync + 'static {
+ /// Create a new [`HttpClient`] with the provided [`ClientOptions`]
+ fn connect(&self, options: &ClientOptions) -> crate::Result<HttpClient>;
+}
+
+/// [`HttpConnector`] using [`reqwest::Client`]
+#[derive(Debug, Default)]
+#[allow(missing_copy_implementations)]
+pub struct ReqwestConnector {}
+
+impl HttpConnector for ReqwestConnector {
+ fn connect(&self, options: &ClientOptions) -> crate::Result<HttpClient> {
+ let client = options.client()?;
Review Comment:
It seems like `reqwest` is still a bit of a special snowflake and has special integration w/ `object_store`. I wonder if it would help to move ALL `reqwest` implementations into a single submodule and make it an optional dependency. Ideally the `reqwest` impls. should also NOT use any `pub(super)`/`pub(crate)`/private methods. This of course could only be enforced if we move the `reqwest`-related code into a new crate, but I until `object_store` doesn't have it's own repo (ref #6183) I think this is going to be too messy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Decouple ObjectStore from Reqwest [arrow-rs]
Posted by "crepererum (via GitHub)" <gi...@apache.org>.
crepererum commented on PR #7183:
URL: https://github.com/apache/arrow-rs/pull/7183#issuecomment-2677966353
Also I think the keyword that is used by other libs for this kind of refactor is called "sans-io", i.e. to make no assumption about the underlying IO layer or framework.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Decouple ObjectStore from Reqwest [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7183:
URL: https://github.com/apache/arrow-rs/pull/7183#discussion_r1967366857
##########
object_store/src/client/retry.rs:
##########
@@ -284,164 +334,97 @@ impl RetryableRequest {
return Ok(r);
}
- let status = r.status();
- let headers = r.headers().clone();
-
- let bytes = r.bytes().await.map_err(|e| Error::Reqwest {
- retries,
- max_retries,
- elapsed: now.elapsed(),
- retry_timeout,
- source: e,
- })?;
-
- let response_body = String::from_utf8_lossy(&bytes);
- debug!("Checking for error in response_body: {}", response_body);
+ let (parts, body) = r.into_parts();
+ let body = match body.text().await {
+ Ok(body) => body,
+ Err(e) => return Err(ctx.err(RequestError::Http(e))),
+ };
- if !body_contains_error(&response_body) {
+ if !body_contains_error(&body) {
// Success response and no error, clone and return response
- let mut success_response = hyper::Response::new(bytes);
- *success_response.status_mut() = status;
- *success_response.headers_mut() = headers;
-
- return Ok(reqwest::Response::from(success_response));
+ return Ok(HttpResponse::from_parts(parts, body.into()));
} else {
// Retry as if this was a 5xx response
- if retries == max_retries || now.elapsed() > retry_timeout {
- return Err(Error::Server {
- body: Some(response_body.into_owned()),
- status,
- });
+ if ctx.exhausted() {
+ return Err(ctx.err(RequestError::Response { body, status }));
}
let sleep = backoff.next();
- retries += 1;
+ ctx.retries += 1;
info!(
"Encountered a response status of {} but body contains Error, backing off for {} seconds, retry {} of {}",
status,
sleep.as_secs_f32(),
- retries,
- max_retries,
+ ctx.retries,
+ ctx.max_retries,
);
tokio::time::sleep(sleep).await;
}
- }
- Ok(r) if r.status() == StatusCode::NOT_MODIFIED => {
- return Err(Error::Client {
- body: None,
- status: StatusCode::NOT_MODIFIED,
- })
- }
- Ok(r) => {
- let is_bare_redirect =
- r.status().is_redirection() && !r.headers().contains_key(LOCATION);
+ } else if status == StatusCode::NOT_MODIFIED {
+ return Err(ctx.err(RequestError::Status { status, body: None }));
+ } else if status.is_redirection() {
+ let is_bare_redirect = !r.headers().contains_key(LOCATION);
return match is_bare_redirect {
- true => Err(Error::BareRedirect),
- // Not actually sure if this is reachable, but here for completeness
- false => Err(Error::Client {
+ true => Err(ctx.err(RequestError::BareRedirect)),
+ false => Err(ctx.err(RequestError::Status {
body: None,
status: r.status(),
- }),
+ })),
};
- }
- Err(e) => {
- let e = sanitize_err(e);
+ } else {
let status = r.status();
- if retries == max_retries
- || now.elapsed() > retry_timeout
+ if ctx.exhausted()
|| !(status.is_server_error()
|| (self.retry_on_conflict && status == StatusCode::CONFLICT))
{
- return Err(match status.is_client_error() {
- true => match r.text().await {
- Ok(body) => Error::Client {
- body: Some(body).filter(|b| !b.is_empty()),
+ let source = match status.is_client_error() {
+ true => match r.into_body().text().await {
+ Ok(body) => RequestError::Status {
status,
+ body: Some(body),
},
- Err(e) => Error::Reqwest {
- retries,
- max_retries,
- elapsed: now.elapsed(),
- retry_timeout,
- source: e,
- },
+ Err(e) => RequestError::Http(e),
},
- false => Error::Reqwest {
- retries,
- max_retries,
- elapsed: now.elapsed(),
- retry_timeout,
- source: e,
- },
- });
- }
+ false => RequestError::Status { status, body: None },
+ };
+ return Err(ctx.err(source));
+ };
let sleep = backoff.next();
- retries += 1;
+ ctx.retries += 1;
info!(
- "Encountered server error, backing off for {} seconds, retry {} of {}: {}",
+ "Encountered server error, backing off for {} seconds, retry {} of {}",
sleep.as_secs_f32(),
- retries,
- max_retries,
- e,
+ ctx.retries,
+ ctx.max_retries,
);
tokio::time::sleep(sleep).await;
}
- },
+ }
Err(e) => {
- let e = sanitize_err(e);
-
- let mut do_retry = false;
- if e.is_connect()
- || e.is_body()
- || (e.is_request() && !e.is_timeout())
- || (is_idempotent && e.is_timeout())
+ // let e = sanitize_err(e);
+
+ let do_retry = match e.kind() {
+ HttpErrorKind::Connect | HttpErrorKind::Request => true, // Request not sent, can retry
+ HttpErrorKind::Timeout => is_idempotent,
+ HttpErrorKind::Unknown
+ | HttpErrorKind::Interrupted
Review Comment:
Note: I think we should retry Interrupted if request is_idempotent
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Decouple ObjectStore from Reqwest [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on code in PR #7183:
URL: https://github.com/apache/arrow-rs/pull/7183#discussion_r1972512325
##########
object_store/src/client/connection.rs:
##########
@@ -0,0 +1,237 @@
+// 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 crate::client::body::{HttpRequest, HttpResponse};
+use crate::client::builder::{HttpRequestBuilder, RequestBuilderError};
+use crate::client::HttpResponseBody;
+use crate::ClientOptions;
+use async_trait::async_trait;
+use http::{Method, Uri};
+use http_body_util::BodyExt;
+use std::error::Error;
+use std::sync::Arc;
+
+/// An HTTP protocol error
+///
+/// Clients should return this when an HTTP request fails to be completed, e.g. because
+/// of a connection issue. This does **not** include HTTP requests that are return
+/// non 2xx Status Codes, as these should instead be returned as an [`HttpResponse`]
+/// with the appropriate status code set.
+#[derive(Debug, thiserror::Error)]
+#[error("HTTP error: {source}")]
+pub struct HttpError {
+ kind: HttpErrorKind,
+ #[source]
+ source: Box<dyn Error + Send + Sync>,
+}
+
+/// Identifies the kind of [`HttpError`]
+///
+/// This is used, among other things, to determine if a request can be retried
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum HttpErrorKind {
+ /// An error occurred whilst connecting to the remote
+ ///
+ /// Will be automatically retried
+ Connect,
+ /// An error occurred whilst making the request
+ ///
+ /// Will be automatically retried
+ Request,
+ /// Request timed out
+ ///
+ /// Will be automatically retried if the request is idempotent
+ Timeout,
+ /// The request was aborted
+ ///
+ /// Will be automatically retried if the request is idempotent
+ Interrupted,
+ /// An error occurred whilst decoding the response
+ ///
+ /// Will not be automatically retried
+ Decode,
+ /// An unknown error occurred
+ ///
+ /// Will not be automatically retried
+ Unknown,
+}
+
+impl HttpError {
+ /// Create a new [`HttpError`] with the optional status code
+ pub fn new<E>(kind: HttpErrorKind, e: E) -> Self
+ where
+ E: Error + Send + Sync + 'static,
+ {
+ Self {
+ kind,
+ source: Box::new(e),
+ }
+ }
+
+ pub(crate) fn reqwest(e: reqwest::Error) -> Self {
+ let mut kind = if e.is_timeout() {
+ HttpErrorKind::Timeout
+ } else if e.is_connect() {
+ HttpErrorKind::Connect
+ } else if e.is_decode() {
+ HttpErrorKind::Decode
+ } else {
+ HttpErrorKind::Unknown
+ };
+
+ // Reqwest error variants aren't great, attempt to refine them
+ let mut source = e.source();
+ while let Some(e) = source {
+ if let Some(e) = e.downcast_ref::<hyper::Error>() {
+ if e.is_closed() || e.is_incomplete_message() || e.is_body_write_aborted() {
+ kind = HttpErrorKind::Request;
+ } else if e.is_timeout() {
+ kind = HttpErrorKind::Timeout;
+ }
+ break;
+ }
+ if let Some(e) = e.downcast_ref::<std::io::Error>() {
+ match e.kind() {
+ std::io::ErrorKind::TimedOut => kind = HttpErrorKind::Timeout,
+ std::io::ErrorKind::ConnectionAborted
+ | std::io::ErrorKind::BrokenPipe
+ | std::io::ErrorKind::UnexpectedEof => kind = HttpErrorKind::Interrupted,
+ _ => {}
+ }
+ break;
+ }
+ source = e.source();
+ }
+ Self {
+ kind,
+ // We strip URL as it will be included by RetryError if not sensitive
+ source: Box::new(e.without_url()),
+ }
+ }
+
+ /// Returns the [`HttpErrorKind`]
+ pub fn kind(&self) -> HttpErrorKind {
+ self.kind
+ }
+}
+
+/// An asynchronous function from a [`HttpRequest`] to a [`HttpResponse`].
+#[async_trait]
+pub trait HttpService: std::fmt::Debug + Send + Sync + 'static {
+ /// Perform [`HttpRequest`] returning [`HttpResponse`]
+ async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError>;
+}
+
+/// An HTTP client
+#[derive(Debug, Clone)]
+pub struct HttpClient(Arc<dyn HttpService>);
+
+impl HttpClient {
+ /// Create a new [`HttpClient`] from an [`HttpService`]
+ pub fn new(service: impl HttpService + 'static) -> Self {
+ Self(Arc::new(service))
+ }
+
+ /// Performs [`HttpRequest`] using this client
+ pub async fn execute(&self, request: HttpRequest) -> Result<HttpResponse, HttpError> {
+ self.0.call(request).await
+ }
+
+ #[allow(unused)]
+ pub(crate) fn get<U>(&self, url: U) -> HttpRequestBuilder
+ where
+ U: TryInto<Uri>,
+ U::Error: Into<RequestBuilderError>,
+ {
+ self.request(Method::GET, url)
+ }
+
+ #[allow(unused)]
+ pub(crate) fn post<U>(&self, url: U) -> HttpRequestBuilder
+ where
+ U: TryInto<Uri>,
+ U::Error: Into<RequestBuilderError>,
+ {
+ self.request(Method::POST, url)
+ }
+
+ #[allow(unused)]
+ pub(crate) fn put<U>(&self, url: U) -> HttpRequestBuilder
+ where
+ U: TryInto<Uri>,
+ U::Error: Into<RequestBuilderError>,
+ {
+ self.request(Method::PUT, url)
+ }
+
+ #[allow(unused)]
+ pub(crate) fn delete<U>(&self, url: U) -> HttpRequestBuilder
+ where
+ U: TryInto<Uri>,
+ U::Error: Into<RequestBuilderError>,
+ {
+ self.request(Method::DELETE, url)
+ }
+
+ pub(crate) fn request<U>(&self, method: Method, url: U) -> HttpRequestBuilder
+ where
+ U: TryInto<Uri>,
+ U::Error: Into<RequestBuilderError>,
+ {
+ HttpRequestBuilder::new(self.clone())
+ .uri(url)
+ .method(method)
+ }
+}
+
+#[async_trait]
+impl HttpService for reqwest::Client {
+ async fn call(&self, req: HttpRequest) -> Result<HttpResponse, HttpError> {
+ let (parts, body) = req.into_parts();
+
+ let url = parts.uri.to_string().parse().unwrap();
+ let mut req = reqwest::Request::new(parts.method, url);
+ *req.headers_mut() = parts.headers;
+ *req.body_mut() = Some(body.into_reqwest());
+
+ let r = self.execute(req).await.map_err(HttpError::reqwest)?;
+ let res: http::Response<reqwest::Body> = r.into();
+ let (parts, body) = res.into_parts();
+
+ let body = HttpResponseBody::new(body.map_err(HttpError::reqwest));
+ Ok(HttpResponse::from_parts(parts, body))
+ }
+}
+
+/// A factory for [`HttpClient`]
+pub trait HttpConnector: std::fmt::Debug + Send + Sync + 'static {
+ /// Create a new [`HttpClient`] with the provided [`ClientOptions`]
+ fn connect(&self, options: &ClientOptions) -> crate::Result<HttpClient>;
+}
+
+/// [`HttpConnector`] using [`reqwest::Client`]
+#[derive(Debug, Default)]
+#[allow(missing_copy_implementations)]
+pub struct ReqwestConnector {}
+
+impl HttpConnector for ReqwestConnector {
+ fn connect(&self, options: &ClientOptions) -> crate::Result<HttpClient> {
+ let client = options.client()?;
Review Comment:
https://github.com/apache/arrow-rs/pull/7203 makes reqwest an optional feature, and shows that the only area it actually leaks into the public API is through some certificate shenanigans on ClientOptions (that can just be gated by the feature).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Decouple ObjectStore from Reqwest [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold merged PR #7183:
URL: https://github.com/apache/arrow-rs/pull/7183
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Decouple ObjectStore from Reqwest [arrow-rs]
Posted by "tustvold (via GitHub)" <gi...@apache.org>.
tustvold commented on PR #7183:
URL: https://github.com/apache/arrow-rs/pull/7183#issuecomment-2677984588
> Is it right to say that instead of reqwest, we now settle on
I think we just depend on `http`, I don't believe we have a dependency on `hyper`
> I wonder if we need the custom service type though or if we could use something like tower
I went back and forth on this, the reason for preferring a custom service was the following:
* Hyper actually defines its own service and doesn't use tower (as tower isn't 1.0) - https://docs.rs/hyper/latest/hyper/service/trait.HttpService.html
* We need a dyn-compatible trait, at which point the existing ecosystem traits become very cumbersome
* Tower is exceptionally confusing, and the `&mut self` requirement is rather obnoxious
* It allows adding Debug + Send + Sync requirements easily
Ultimately defining a very simple async_trait in terms of `http` and custom body implementations, seemed simpler and unlikely to paint us into a corner
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org