upgrade to reqwest 0.12

Signed-off-by: tison <[email protected]>

Cargo.toml

@@ -146,7 +146,7 @@ raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.8"
 regex = "1.8"
 regex-automata = { version = "0.4" }
-reqwest = { version = "0.11", default-features = false, features = [
+reqwest = { version = "0.12", default-features = false, features = [
     "json",
     "rustls-tls-native-roots",
     "stream",

src/datanode/src/store.rs

@@ -20,7 +20,8 @@ mod gcs;
 mod oss;
 mod s3;
 
-use std::path;
+use std::{env, path};
+use std::time::Duration;
 
 use common_base::readable_size::ReadableSize;
 use common_telemetry::info;
@@ -136,38 +137,35 @@ pub(crate) fn clean_temp_dir(dir: &str) -> Result<()> {
     Ok(())
 }
 
-/// FIXME: we need to use reqwest 0.12 here.
 pub(crate) fn build_http_client() -> Result<HttpClient> {
-    // let http_builder = {
-    //     let mut builder = reqwest::ClientBuilder::new();
-    //
-    //     // Pool max idle per host controls connection pool size.
-    //     // Default to no limit, set to `0` for disable it.
-    //     let pool_max_idle_per_host = env::var("_GREPTIMEDB_HTTP_POOL_MAX_IDLE_PER_HOST")
-    //         .ok()
-    //         .and_then(|v| v.parse::<usize>().ok())
-    //         .unwrap_or(usize::MAX);
-    //     builder = builder.pool_max_idle_per_host(pool_max_idle_per_host);
-    //
-    //     // Connect timeout default to 30s.
-    //     let connect_timeout = env::var("_GREPTIMEDB_HTTP_CONNECT_TIMEOUT")
-    //         .ok()
-    //         .and_then(|v| v.parse::<u64>().ok())
-    //         .unwrap_or(30);
-    //     builder = builder.connect_timeout(Duration::from_secs(connect_timeout));
-    //
-    //     // Pool connection idle timeout default to 90s.
-    //     let idle_timeout = env::var("_GREPTIMEDB_HTTP_POOL_IDLE_TIMEOUT")
-    //         .ok()
-    //         .and_then(|v| v.parse::<u64>().ok())
-    //         .unwrap_or(90);
-    //
-    //     builder = builder.pool_idle_timeout(Duration::from_secs(idle_timeout));
-    //
-    //     builder
-    // };
-    //
-    // HttpClient::build(http_builder).context(error::InitBackendSnafu)
-
-    HttpClient::new().context(error::InitBackendSnafu)
+    let http_builder = {
+        let mut builder = reqwest::ClientBuilder::new();
+
+        // Pool max idle per host controls connection pool size.
+        // Default to no limit, set to `0` for disable it.
+        let pool_max_idle_per_host = env::var("_GREPTIMEDB_HTTP_POOL_MAX_IDLE_PER_HOST")
+            .ok()
+            .and_then(|v| v.parse::<usize>().ok())
+            .unwrap_or(usize::MAX);
+        builder = builder.pool_max_idle_per_host(pool_max_idle_per_host);
+
+        // Connect timeout default to 30s.
+        let connect_timeout = env::var("_GREPTIMEDB_HTTP_CONNECT_TIMEOUT")
+            .ok()
+            .and_then(|v| v.parse::<u64>().ok())
+            .unwrap_or(30);
+        builder = builder.connect_timeout(Duration::from_secs(connect_timeout));
+
+        // Pool connection idle timeout default to 90s.
+        let idle_timeout = env::var("_GREPTIMEDB_HTTP_POOL_IDLE_TIMEOUT")
+            .ok()
+            .and_then(|v| v.parse::<u64>().ok())
+            .unwrap_or(90);
+
+        builder = builder.pool_idle_timeout(Duration::from_secs(idle_timeout));
+
+        builder
+    };
+
+    HttpClient::build(http_builder).context(error::InitBackendSnafu)
 }

src/servers/src/export_metrics.rs

@@ -16,14 +16,12 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
 
-use axum::http::HeaderValue;
 use common_base::Plugins;
 use common_telemetry::metric::{convert_metric_to_write_request, MetricFilter};
 use common_telemetry::{error, info};
 use common_time::Timestamp;
-use hyper::HeaderMap;
 use prost::Message;
-use reqwest::header::HeaderName;
+use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
 use serde::{Deserialize, Serialize};
 use session::context::QueryContextBuilder;
 use snafu::{ensure, ResultExt};
@@ -115,7 +113,7 @@ impl ExportMetricsTask {
                 }
             );
         }
-        let mut headers = reqwest::header::HeaderMap::new();
+        let mut headers = HeaderMap::new();
         if let Some(remote_write) = &config.remote_write {
             ensure!(
                 !remote_write.url.is_empty(),

src/servers/src/http/greptime_result_v1.rs

@@ -14,10 +14,10 @@
 
 use std::collections::HashMap;
 
+use axum::headers::HeaderValue;
 use axum::response::{IntoResponse, Response};
 use axum::Json;
 use common_query::Output;
-use reqwest::header::HeaderValue;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;

src/servers/src/http/test_helpers.rs

@@ -32,6 +32,7 @@
 
 use std::convert::TryFrom;
 use std::net::{SocketAddr, TcpListener};
+use std::str::FromStr;
 
 use axum::body::HttpBody;
 use axum::BoxError;
@@ -169,7 +170,15 @@ impl RequestBuilder {
         HeaderValue: TryFrom<V>,
         <HeaderValue as TryFrom<V>>::Error: Into<http::Error>,
     {
+        // TODO(tisonkun): revert once http bump to 1.x
+        let key: HeaderName = key.try_into().map_err(Into::into).unwrap();
+        let key = reqwest::header::HeaderName::from_bytes(key.as_ref()).unwrap();
+
+        let value: HeaderValue = value.try_into().map_err(Into::into).unwrap();
+        let value = reqwest::header::HeaderValue::from_bytes(value.as_bytes()).unwrap();
+
         self.builder = self.builder.header(key, value);
+
         self
     }
 
@@ -210,12 +219,19 @@ impl TestResponse {
 
     /// Get the response status.
     pub fn status(&self) -> StatusCode {
-        self.response.status()
+        StatusCode::from_u16(self.response.status().as_u16()).unwrap()
     }
 
     /// Get the response headers.
-    pub fn headers(&self) -> &http::HeaderMap {
-        self.response.headers()
+    pub fn headers(&self) -> http::HeaderMap {
+        // TODO(tisonkun): revert once http bump to 1.x
+        let mut headers = http::HeaderMap::new();
+        for (key, value) in self.response.headers() {
+            let key = HeaderName::from_str(key.as_str()).unwrap();
+            let value = HeaderValue::from_bytes(value.as_bytes()).unwrap();
+            headers.insert(key, value);
+        }
+        headers
     }
 
     /// Get the response in chunks.