feat: support custom Azure Blob Storage endpoints for sovereign clouds

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- Azure Blob Storage: support custom endpoints via `endpoint` and `endpoint_suffix` configuration options for sovereign clouds (#6624)
 
 ### Fixed
 - (Jaeger) Query resource attributes when Jaeger request carries tags

docs/configuration/storage-config.md

@@ -110,13 +110,17 @@ storage:
 | --- | --- | --- |
 | `account` | The Azure storage account name. | |
 | `access_key` | The Azure storage account access key. | |
+| `endpoint` | Custom blob service endpoint URL. | SDK default (`https://<account>.blob.core.windows.net`) |
+| `endpoint_suffix` | Blob service endpoint suffix for sovereign clouds. Ignored when `endpoint` is set. | SDK default (`core.windows.net`) |
 
 #### Environment variables
 
 | Env variable | Description |
 | --- | --- |
 | `QW_AZURE_STORAGE_ACCOUNT` | Azure Blob Storage account name. |
 | `QW_AZURE_STORAGE_ACCESS_KEY` | Azure Blob Storage account access key. |
+| `QW_AZURE_ENDPOINT` | Custom blob service endpoint URL. |
+| `QW_AZURE_ENDPOINT_SUFFIX` | Blob service endpoint suffix for sovereign clouds. |
 
 Example of a storage configuration for Azure in YAML format:
 
@@ -127,6 +131,24 @@ storage:
     access_key: your-azure-access-key
 ```
 
+Example for Azure US Government:
+
+```yaml
+storage:
+  azure:
+    account: your-azure-account-name
+    endpoint_suffix: core.usgovcloudapi.net
+```
+
+Example for Azure China:
+
+```yaml
+storage:
+  azure:
+    account: your-azure-account-name
+    endpoint_suffix: core.chinacloudapi.cn
+```
+
 ## Storage configuration examples for various object storage providers
 
 ### Garage

quickwit/quickwit-config/src/storage_config.rs

@@ -18,7 +18,7 @@ use std::{env, fmt};
 
 use anyhow::ensure;
 use itertools::Itertools;
-use quickwit_common::get_bool_from_env;
+use quickwit_common::{get_bool_from_env, get_from_env_opt};
 use serde::{Deserialize, Serialize};
 use serde_with::{EnumMap, serde_as};
 
@@ -289,13 +289,25 @@ pub struct AzureStorageConfig {
     #[serde(default)]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub access_key: Option<String>,
+    /// Custom blob service endpoint URL, e.g. `https://myaccount.blob.core.usgovcloudapi.net`.
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub endpoint: Option<String>,
+    /// Blob service endpoint suffix for sovereign clouds, e.g. `core.usgovcloudapi.net`.
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub endpoint_suffix: Option<String>,
 }
 
 impl AzureStorageConfig {
     pub const AZURE_STORAGE_ACCOUNT_ENV_VAR: &'static str = "QW_AZURE_STORAGE_ACCOUNT";
 
     pub const AZURE_STORAGE_ACCESS_KEY_ENV_VAR: &'static str = "QW_AZURE_STORAGE_ACCESS_KEY";
 
+    pub const AZURE_ENDPOINT_ENV_VAR: &'static str = "QW_AZURE_ENDPOINT";
+
+    pub const AZURE_ENDPOINT_SUFFIX_ENV_VAR: &'static str = "QW_AZURE_ENDPOINT_SUFFIX";
+
     /// Redacts the access key.
     pub fn redact(&mut self) {
         if let Some(access_key) = self.access_key.as_mut() {
@@ -306,18 +318,45 @@ impl AzureStorageConfig {
     /// Attempts to find the storage account name in the environment variable
     /// `QW_AZURE_STORAGE_ACCOUNT` or node config.
     pub fn resolve_account_name(&self) -> Option<String> {
-        env::var(Self::AZURE_STORAGE_ACCOUNT_ENV_VAR)
-            .ok()
+        get_from_env_opt(Self::AZURE_STORAGE_ACCOUNT_ENV_VAR, false)
             .or_else(|| self.account_name.clone())
     }
 
     /// Attempts to find the storage account access key in the environment variable
     /// `QW_AZURE_STORAGE_ACCESS_KEY` or node config.
     pub fn resolve_access_key(&self) -> Option<String> {
-        env::var(Self::AZURE_STORAGE_ACCESS_KEY_ENV_VAR)
-            .ok()
+        get_from_env_opt(Self::AZURE_STORAGE_ACCESS_KEY_ENV_VAR, true)
             .or_else(|| self.access_key.clone())
     }
+
+    /// Attempts to find the blob service endpoint URL in the environment variable
+    /// `QW_AZURE_ENDPOINT` or node config.
+    pub fn endpoint(&self) -> Option<String> {
+        get_from_env_opt(Self::AZURE_ENDPOINT_ENV_VAR, false).or_else(|| self.endpoint.clone())
+    }
+
+    /// Attempts to find the blob service endpoint suffix in the environment variable
+    /// `QW_AZURE_ENDPOINT_SUFFIX` or node config.
+    pub fn endpoint_suffix(&self) -> Option<String> {
+        get_from_env_opt(Self::AZURE_ENDPOINT_SUFFIX_ENV_VAR, false)
+            .or_else(|| self.endpoint_suffix.clone())
+    }
+
+    /// Builds the blob service URI when `endpoint` or `endpoint_suffix` is configured.
+    ///
+    /// When both are set, `endpoint` takes precedence.
+    pub fn resolve_blob_service_uri(&self, account_name: &str) -> Option<String> {
+        if let Some(endpoint) = self.endpoint() {
+            return Some(endpoint);
+        }
+        let endpoint_suffix = self.endpoint_suffix()?;
+        let uri = if endpoint_suffix.starts_with("blob.") {
+            format!("https://{account_name}.{endpoint_suffix}")
+        } else {
+            format!("https://{account_name}.blob.{endpoint_suffix}")
+        };
+        Some(uri)
+    }
 }
 
 impl fmt::Debug for AzureStorageConfig {
@@ -328,6 +367,8 @@ impl fmt::Debug for AzureStorageConfig {
                 "access_key",
                 &self.access_key.as_ref().map(|_| "***redacted***"),
             )
+            .field("endpoint", &self.endpoint)
+            .field("endpoint_suffix", &self.endpoint_suffix)
             .finish()
     }
 }
@@ -630,11 +671,64 @@ mod tests {
             let expected_azure_config = AzureStorageConfig {
                 account_name: Some("test-account".to_string()),
                 access_key: Some("test-access-key".to_string()),
+                ..Default::default()
+            };
+            assert_eq!(azure_storage_config, expected_azure_config);
+        }
+        {
+            let azure_storage_config_yaml = r#"
+                account: test-account
+                endpoint: https://test-account.blob.core.usgovcloudapi.net
+                endpoint_suffix: core.chinacloudapi.cn
+            "#;
+            let azure_storage_config: AzureStorageConfig =
+                serde_yaml::from_str(azure_storage_config_yaml).unwrap();
+
+            let expected_azure_config = AzureStorageConfig {
+                account_name: Some("test-account".to_string()),
+                endpoint: Some("https://test-account.blob.core.usgovcloudapi.net".to_string()),
+                endpoint_suffix: Some("core.chinacloudapi.cn".to_string()),
+                ..Default::default()
             };
             assert_eq!(azure_storage_config, expected_azure_config);
         }
     }
 
+    #[test]
+    fn test_storage_azure_config_resolve_blob_service_uri() {
+        let config = AzureStorageConfig {
+            account_name: Some("my-account".to_string()),
+            endpoint: Some("https://custom.example.com".to_string()),
+            endpoint_suffix: Some("core.usgovcloudapi.net".to_string()),
+            ..Default::default()
+        };
+        assert_eq!(
+            config.resolve_blob_service_uri("my-account").as_deref(),
+            Some("https://custom.example.com")
+        );
+
+        let config = AzureStorageConfig {
+            endpoint_suffix: Some("core.usgovcloudapi.net".to_string()),
+            ..Default::default()
+        };
+        assert_eq!(
+            config.resolve_blob_service_uri("my-account").as_deref(),
+            Some("https://my-account.blob.core.usgovcloudapi.net")
+        );
+
+        let config = AzureStorageConfig {
+            endpoint_suffix: Some("blob.core.chinacloudapi.cn".to_string()),
+            ..Default::default()
+        };
+        assert_eq!(
+            config.resolve_blob_service_uri("my-account").as_deref(),
+            Some("https://my-account.blob.core.chinacloudapi.cn")
+        );
+
+        let config = AzureStorageConfig::default();
+        assert!(config.resolve_blob_service_uri("my-account").is_none());
+    }
+
     #[test]
     fn test_storage_google_config_serde() {
         {

quickwit/quickwit-storage/src/object_storage/azure_blob_storage.rs

@@ -22,8 +22,8 @@ use std::{fmt, io};
 use async_trait::async_trait;
 use azure_core::error::ErrorKind;
 use azure_core::{Pageable, StatusCode};
-use azure_storage::Error as AzureError;
 use azure_storage::prelude::*;
+use azure_storage::{CloudLocation, Error as AzureError};
 use azure_storage_blobs::blob::operations::GetBlobResponse;
 use azure_storage_blobs::prelude::*;
 use bytes::{Bytes, BytesMut};
@@ -42,7 +42,7 @@ use thiserror::Error;
 use tokio::io::{AsyncRead, AsyncWriteExt, BufReader};
 use tokio_util::compat::FuturesAsyncReadCompatExt;
 use tokio_util::io::StreamReader;
-use tracing::{instrument, warn};
+use tracing::{info, instrument, warn};
 
 use crate::debouncer::DebouncedStorage;
 use crate::metrics::object_storage_get_slice_in_flight_guards;
@@ -100,11 +100,16 @@ impl AzureBlobStorage {
     pub fn new(
         storage_account_name: String,
         storage_credentials: StorageCredentials,
+        blob_service_uri: Option<String>,
         uri: Uri,
         container_name: String,
     ) -> Self {
-        let container_client = BlobServiceClient::new(storage_account_name, storage_credentials)
-            .container_client(container_name);
+        let container_client = build_container_client(
+            storage_account_name,
+            storage_credentials,
+            blob_service_uri,
+            container_name,
+        );
         Self {
             container_client,
             uri,
@@ -192,9 +197,11 @@ impl AzureBlobStorage {
             let message = format!("failed to extract container name from Azure URI `{uri}`");
             StorageResolverError::InvalidUri(message)
         })?;
+        let blob_service_uri = azure_storage_config.resolve_blob_service_uri(&storage_account_name);
         let azure_blob_storage = AzureBlobStorage::new(
             storage_account_name,
             storage_credentials,
+            blob_service_uri,
             uri.clone(),
             container_name,
         );
@@ -556,6 +563,25 @@ async fn extract_range_data_and_hash(
     Ok((data, hash))
 }
 
+fn build_container_client(
+    storage_account_name: String,
+    storage_credentials: StorageCredentials,
+    blob_service_uri: Option<String>,
+    container_name: String,
+) -> ContainerClient {
+    let mut builder = ClientBuilder::new(storage_account_name.clone(), storage_credentials);
+    if let Some(uri) = blob_service_uri {
+        info!(endpoint=%uri, "using Azure blob storage endpoint defined in storage config or environment variable");
+        builder = builder.cloud_location(CloudLocation::Custom {
+            account: storage_account_name,
+            uri,
+        });
+    }
+    builder
+        .blob_service_client()
+        .container_client(container_name)
+}
+
 pub fn parse_azure_uri(uri: &Uri) -> Option<(String, PathBuf)> {
     // Ex: azure://container/prefix.
     static URI_PTN: LazyLock<Regex> = LazyLock::new(|| {