refactor: make fuse_amend implementation more convenient for on-premise deployments (#17147)

* tweak cargo.toml

* use native_tls in fuse_amend

* sync Cargo.lock

* add settings

Author: dantengsky

Cargo.lock

@@ -234,11 +234,11 @@ async-compression = { git = "https://github.com/datafuse-extras/async-compressio
     "all-algorithms",
 ] }
 async-recursion = "1.1.1"
-async-std = "1.12"
 async-stream = "0.3.3"
 async-trait = { version = "0.1.77" }
-aws-config = "1.1.7"
+aws-config = { version = "1.1.7", features = ["behavior-version-latest"] }
 aws-sdk-s3 = "1.17.0"
+aws-smithy-runtime = { version = "1.6.2", default-features = false }
 backoff = "0.4" # FIXME: use backon to replace this.
 backon = "1"
 backtrace = "0.3.73"
@@ -318,7 +318,9 @@ hostname = "0.3.1"
 http = "1"
 humantime = "2.1.0"
 hyper = "1"
+hyper-tls = "0.5.0"
 hyper-util = { version = "0.1.9", features = ["client", "client-legacy", "tokio", "service"] }
+hyper_v014 = { package = "hyper", version = "0.14" }
 iceberg = { version = "0.3.0", git = "https://github.com/Xuanwo/iceberg-rust/", rev = "01d706a1" }
 iceberg-catalog-glue = { version = "0.3.0", git = "https://github.com/Xuanwo/iceberg-rust/", rev = "01d706a1" }
 iceberg-catalog-hms = { version = "0.3.0", git = "https://github.com/Xuanwo/iceberg-rust/", rev = "01d706a1" }

Cargo.toml

@@ -18,6 +18,7 @@ anyerror = { workspace = true }
 hickory-resolver = { workspace = true }
 hyper = { workspace = true }
 hyper-util = { workspace = true }
+hyper_v014 = { workspace = true }
 jwt-simple = { workspace = true }
 log = { workspace = true }
 serde = { workspace = true }

src/common/grpc/Cargo.toml

@@ -32,6 +32,7 @@ use hickory_resolver::TokioAsyncResolver;
 use hyper::Uri;
 use hyper_util::client::legacy::connect::dns::Name;
 use hyper_util::client::legacy::connect::HttpConnector;
+use hyper_v014::client::connect::dns::Name as Hyperv014Name;
 use log::info;
 use serde::Deserialize;
 use serde::Serialize;
@@ -109,6 +110,31 @@ impl tower_service::Service<Name> for DNSService {
     }
 }
 
+// resolve Name of hyper version 0.14
+impl tower_service::Service<Hyperv014Name> for DNSService {
+    type Response = DNSServiceAddrs;
+    type Error = ErrorCode;
+    type Future = DNSServiceFuture;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<std::result::Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, name: Hyperv014Name) -> Self::Future {
+        let blocking = runtime::spawn(async move {
+            let resolver = DNSResolver::instance()?;
+            match resolver.resolve(name.to_string()).await {
+                Err(err) => Err(err),
+                Ok(addrs) => Ok(DNSServiceAddrs {
+                    inner: addrs.into_iter(),
+                }),
+            }
+        });
+
+        DNSServiceFuture { inner: blocking }
+    }
+}
+
 pub struct DNSServiceFuture {
     inner: JoinHandle<Result<DNSServiceAddrs>>,
 }

src/common/grpc/src/dns_resolver.rs

@@ -14,6 +14,7 @@ test = true
 [dependencies]
 async-backtrace = { workspace = true }
 async-trait = { workspace = true }
+aws-smithy-runtime = { workspace = true }
 base64 = { workspace = true }
 chrono = { workspace = true }
 chrono-tz = { workspace = true }
@@ -25,6 +26,7 @@ databend-common-config = { workspace = true }
 databend-common-exception = { workspace = true }
 databend-common-expression = { workspace = true }
 databend-common-functions = { workspace = true }
+databend-common-grpc = { workspace = true }
 databend-common-io = { workspace = true }
 databend-common-license = { workspace = true }
 databend-common-meta-api = { workspace = true }
@@ -61,13 +63,15 @@ databend-storages-common-table-meta = { workspace = true }
 derive-visitor = { workspace = true }
 futures = { workspace = true }
 futures-util = { workspace = true }
+hyper-tls = { workspace = true }
+hyper_v014 = { workspace = true }
 jwt-simple = { workspace = true }
 log = { workspace = true }
 opendal = { workspace = true }
 tempfile = { workspace = true }
 
 # aws sdk
-aws-config = { workspace = true, features = ["behavior-version-latest"] }
+aws-config = { workspace = true }
 aws-sdk-s3 = { workspace = true }
 
 [dev-dependencies]

src/query/ee/Cargo.toml

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::env;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -22,13 +23,16 @@ use aws_sdk_s3::config::Credentials;
 use aws_sdk_s3::config::Region;
 use aws_sdk_s3::config::SharedCredentialsProvider;
 use aws_sdk_s3::Client;
+use aws_smithy_runtime::client::http::hyper_014::HyperClientBuilder;
 use databend_common_base::base::tokio::io::AsyncReadExt;
 use databend_common_base::base::GlobalInstance;
 use databend_common_catalog::table::Table;
+use databend_common_catalog::table_context::TableContext;
 use databend_common_config::GlobalConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::TableSchema;
+use databend_common_grpc::DNSService;
 use databend_common_meta_app::schema::TableInfo;
 use databend_common_meta_app::storage::StorageParams;
 use databend_common_storages_fuse::io::MetaReaders;
@@ -38,6 +42,7 @@ use databend_enterprise_fail_safe::FailSafeHandlerWrapper;
 use databend_storages_common_cache::LoadParams;
 use databend_storages_common_table_meta::meta::CompactSegmentInfo;
 use databend_storages_common_table_meta::meta::Location;
+use hyper_tls::HttpsConnector;
 use log::info;
 use log::warn;
 use opendal::ErrorKind;
@@ -49,7 +54,11 @@ impl RealFailSafeHandler {}
 
 #[async_trait::async_trait]
 impl FailSafeHandler for RealFailSafeHandler {
-    async fn recover_table_data(&self, table_info: TableInfo) -> Result<()> {
+    async fn recover_table_data(
+        &self,
+        ctx: Arc<dyn TableContext>,
+        table_info: TableInfo,
+    ) -> Result<()> {
         let storage_params = match &table_info.meta.storage_params {
             // External or attached table.
             Some(sp) => sp.clone(),
@@ -62,7 +71,7 @@ impl FailSafeHandler for RealFailSafeHandler {
 
         let fuse_table = FuseTable::do_create(table_info)?;
 
-        let amender = Amender::try_new(storage_params).await?;
+        let amender = Amender::try_new(ctx, storage_params).await?;
 
         amender.recover_snapshot(fuse_table).await?;
 
@@ -86,7 +95,7 @@ struct Amender {
 }
 
 impl Amender {
-    async fn try_new(storage_param: StorageParams) -> Result<Self> {
+    async fn try_new(ctx: Arc<dyn TableContext>, storage_param: StorageParams) -> Result<Self> {
         // TODO
         // - replace client with opendal operator
         // - supports other storage types
@@ -116,18 +125,53 @@ impl Amender {
                 .connect_timeout(Duration::from_secs(3))
                 .build();
 
+            let tls_connector = {
+                let mut builder = hyper_tls::native_tls::TlsConnector::builder();
+                let allow_invalid_cert = ctx
+                    .get_settings()
+                    .get_premise_deploy_danger_amend_accept_invalid_cert()?;
+                if allow_invalid_cert {
+                    info!("allows invalid cert, and accepts invalid hostnames in cert validation");
+                    builder.danger_accept_invalid_certs(true);
+                    builder.danger_accept_invalid_hostnames(true);
+                };
+                builder
+                    .build()
+                    .unwrap_or_else(|e| panic!("error while creating TLS connector: {}", e))
+            };
+
+            let mut http = hyper_v014::client::HttpConnector::new_with_resolver(DNSService {});
+            // also allows http connection
+            http.enforce_http(false);
+
+            let connect_timeout = env::var("_DATABEND_INTERNAL_CONNECT_TIMEOUT")
+                .ok()
+                .and_then(|v| v.parse::<u64>().ok())
+                .unwrap_or(30);
+            http.set_connect_timeout(Some(Duration::from_secs(connect_timeout)));
+
+            let conn = HttpsConnector::from((http, tls_connector.into()));
+
             let config = aws_config::from_env()
                 .region(region_provider)
                 .endpoint_url(s3_config.endpoint_url)
                 .credentials_provider(SharedCredentialsProvider::new(base_credentials))
                 .retry_config(retry_config)
                 .timeout_config(timeout_config)
+                .http_client(HyperClientBuilder::new().build(conn))
                 .load()
                 .await;
 
+            let force_path_style = ctx
+                .get_settings()
+                .get_premise_deploy_amend_force_path_style()?;
+            let sdk_config = aws_sdk_s3::config::Builder::from(&config)
+                .force_path_style(force_path_style)
+                .build();
+
             let root = s3_config.root;
             let bucket = s3_config.bucket;
-            let client = Client::new(&config);
+            let client = Client::from_conf(sdk_config);
 
             Ok(Self {
                 client,
@@ -286,7 +330,7 @@ impl Amender {
             .send()
             .await
             .map_err(|e| {
-                ErrorCode::StorageOther(format!("failed to list object versions. {}", e))
+                ErrorCode::StorageOther(format!("failed to list object versions. {:?}", e))
             })?;
 
         // find the latest version

src/query/ee/src/fail_safe/handler.rs

@@ -14,6 +14,7 @@ test = true
 async-backtrace = { workspace = true }
 async-trait = { workspace = true }
 databend-common-base = { workspace = true }
+databend-common-catalog = { workspace = true }
 databend-common-exception = { workspace = true }
 databend-common-meta-app = { workspace = true }
 

src/query/ee_features/fail_safe/Cargo.toml

@@ -15,12 +15,17 @@
 use std::sync::Arc;
 
 use databend_common_base::base::GlobalInstance;
+use databend_common_catalog::table_context::TableContext;
 use databend_common_exception::Result;
 use databend_common_meta_app::schema::TableInfo;
 
 #[async_trait::async_trait]
 pub trait FailSafeHandler: Sync + Send {
-    async fn recover_table_data(&self, table_info: TableInfo) -> Result<()>;
+    async fn recover_table_data(
+        &self,
+        ctx: Arc<dyn TableContext>,
+        table_info: TableInfo,
+    ) -> Result<()>;
 }
 
 pub struct FailSafeHandlerWrapper {
@@ -33,8 +38,8 @@ impl FailSafeHandlerWrapper {
     }
 
     #[async_backtrace::framed]
-    pub async fn recover(&self, table_info: TableInfo) -> Result<()> {
-        self.handler.recover_table_data(table_info).await
+    pub async fn recover(&self, ctx: Arc<dyn TableContext>, table_info: TableInfo) -> Result<()> {
+        self.handler.recover_table_data(ctx, table_info).await
     }
 }
 

src/query/ee_features/fail_safe/src/handler.rs

@@ -1151,6 +1151,20 @@ impl DefaultSettings {
                     scope: SettingScope::Both,
                     range: Some(SettingRange::Numeric(0..=u64::MAX)),
                 }),
+                ("premise_deploy_danger_amend_accept_invalid_cert", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(0),
+                    desc: "Setting this to a non-zero value will allow `fuse_amend` to accept invalid TLS certificates. For diagnostic purposes only, Be very cautious before setting this to a non-zero value. If you're unsure, leave it unchanged.",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(0..=1)),
+                }),
+                ("premise_deploy_amend_force_path_style", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(1),
+                    desc: "Setting this to a non-zero value will let `fuse_amend` use path style uri while accessing s3-compatible storage service.",
+                    mode: SettingMode::Both,
+                    scope: SettingScope::Both,
+                    range: Some(SettingRange::Numeric(0..=1)),
+                }),
             ]);
 
             Ok(Arc::new(DefaultSettings {

src/query/settings/src/settings_default.rs

@@ -848,4 +848,12 @@ impl Settings {
         let v = self.try_get_u64("stream_consume_batch_size_hint")?;
         Ok(if v == 0 { None } else { Some(v) })
     }
+
+    pub fn get_premise_deploy_danger_amend_accept_invalid_cert(&self) -> Result<bool> {
+        Ok(self.try_get_u64("premise_deploy_danger_amend_accept_invalid_cert")? != 0)
+    }
+
+    pub fn get_premise_deploy_amend_force_path_style(&self) -> Result<bool> {
+        Ok(self.try_get_u64("premise_deploy_amend_force_path_style")? != 0)
+    }
 }

src/query/settings/src/settings_getter_setter.rs

@@ -89,7 +89,7 @@ impl SimpleTableFunc for FuseAmendTable {
         })?;
 
         self.fail_safe_handler
-            .recover(tbl.table_info.clone())
+            .recover(ctx.clone(), tbl.table_info.clone())
             .await?;
 
         let col: Vec<String> = vec!["Ok".to_owned()];