feat: add X-DATABEND-TENANT in sharing request

Author: ZhiHanZ

src/common/storage/src/config.rs

@@ -500,17 +500,20 @@ impl Debug for StorageRedisConfig {
 pub struct ShareTableConfig {
     pub share_endpoint_address: Option<String>,
     pub share_endpoint_token: RefreshableToken,
+    pub requester_tenant_id: String,
 }
 
 impl ShareTableConfig {
     pub fn init(
         share_endpoint_address: &str,
         token_file: &str,
+        requester_tenant_id: String,
         default_token: String,
     ) -> common_exception::Result<()> {
         GlobalInstance::set(Self::try_create(
             share_endpoint_address,
             token_file,
+            requester_tenant_id,
             default_token,
         )?);
 
@@ -520,6 +523,7 @@ impl ShareTableConfig {
     pub fn try_create(
         share_endpoint_address: &str,
         token_file: &str,
+        requester_tenant_id: String,
         default_token: String,
     ) -> common_exception::Result<ShareTableConfig> {
         let share_endpoint_address = if share_endpoint_address.is_empty() {
@@ -537,6 +541,7 @@ impl ShareTableConfig {
         Ok(ShareTableConfig {
             share_endpoint_address,
             share_endpoint_token,
+            requester_tenant_id,
         })
     }
 
@@ -548,6 +553,10 @@ impl ShareTableConfig {
         ShareTableConfig::instance().share_endpoint_token
     }
 
+    pub fn requester_tenant_id() -> String {
+        ShareTableConfig::instance().requester_tenant_id
+    }
+
     pub fn instance() -> ShareTableConfig {
         GlobalInstance::get()
     }

src/query/service/src/global_services.rs

@@ -71,6 +71,7 @@ impl GlobalServices {
             &config.query.share_endpoint_address,
             &config.query.share_endpoint_auth_token_file,
             config.query.tenant_id.clone(),
+            config.query.tenant_id.clone(),
         )?;
 
         CacheManager::init(&config.query)?;

src/query/sharing/src/layer.rs

@@ -56,6 +56,7 @@ use crate::SharedSigner;
 pub fn create_share_table_operator(
     share_endpoint_address: Option<String>,
     share_endpoint_token: RefreshableToken,
+    requester_tenant_id: String,
     share_tenant_id: &str,
     share_name: &str,
     table_name: &str,
@@ -68,6 +69,7 @@ pub fn create_share_table_operator(
                     share_endpoint_address, share_tenant_id, share_name, table_name
                 ),
                 share_endpoint_token,
+                requester_tenant_id,
             );
             Operator::new(apply_wrapper(SharedAccessor {
                 signer,

src/query/sharing/src/signer.rs

@@ -33,6 +33,8 @@ use opendal::raw::HttpClient;
 use opendal::raw::Operation;
 use opendal::raw::PresignedRequest;
 
+const TENANT_HEADER: &str = "X-DATABEND-TENANT";
+
 /// SharedSigner is used to track presign request, and it's response.
 ///
 /// There is an internal cache about presign request. Getting an expired
@@ -43,6 +45,7 @@ pub struct SharedSigner {
     cache: Cache<PresignRequest, PresignedRequest>,
     client: HttpClient,
     token: RefreshableToken,
+    requester_tenant_id: String,
 }
 
 impl Debug for SharedSigner {
@@ -55,7 +58,7 @@ impl Debug for SharedSigner {
 
 impl SharedSigner {
     /// Create a new SharedSigner.
-    pub fn new(endpoint: &str, token: RefreshableToken) -> Self {
+    pub fn new(endpoint: &str, token: RefreshableToken, req_tenant_id: String) -> Self {
         let cache = Cache::builder()
             // Databend Cloud Presign will expire after 3600s (1 hour).
             // We will expire them 10 minutes before to avoid edge cases.
@@ -67,6 +70,7 @@ impl SharedSigner {
             cache,
             client: HttpClient::new(),
             token,
+            requester_tenant_id: req_tenant_id,
         }
     }
 
@@ -154,6 +158,7 @@ impl SharedSigner {
             .uri(&self.endpoint)
             .header(AUTHORIZATION, auth)
             .header(CONTENT_LENGTH, bs.len())
+            .header(TENANT_HEADER, self.requester_tenant_id.clone())
             .body(AsyncBody::Bytes(bs))?;
         let resp = self.client.send_async(req).await?;
         let bs = resp.into_body().bytes().await?;

src/query/storages/fuse/fuse/src/fuse_table.rs

@@ -98,6 +98,7 @@ impl FuseTable {
             Some(ref from_share) => create_share_table_operator(
                 ShareTableConfig::share_endpoint_address(),
                 ShareTableConfig::share_endpoint_token(),
+                ShareTableConfig::requester_tenant_id(),
                 &from_share.tenant,
                 &from_share.share_name,
                 &table_info.name,