[object_store] Propagate env vars as object store client options (apa…

…che#6334)

* [object_store] Propagate env vars as object store client options

* [object_store] Include the missing variants in the FromStr implementation of ClientConfigKey

* cargo fmt

object_store/src/aws/builder.rs

@@ -402,7 +402,7 @@ impl FromStr for AmazonS3ConfigKey {
             "aws_sse_customer_key_base64" => Ok(Self::Encryption(
                 S3EncryptionConfigKey::CustomerEncryptionKey,
             )),
-            _ => match s.parse() {
+            _ => match s.strip_prefix("aws_").unwrap_or(s).parse() {
                 Ok(key) => Ok(Self::Client(key)),
                 Err(_) => Err(Error::UnknownConfigurationKey { key: s.into() }.into()),
             },
@@ -1455,4 +1455,17 @@ mod tests {
             assert_eq!(parse_bucket_az(bucket), expected)
         }
     }
+
+    #[test]
+    fn aws_test_client_opts() {
+        let key = "AWS_PROXY_URL";
+        if let Ok(config_key) = key.to_ascii_lowercase().parse() {
+            assert_eq!(
+                AmazonS3ConfigKey::Client(ClientConfigKey::ProxyUrl),
+                config_key
+            );
+        } else {
+            panic!("{} not propagated as ClientConfigKey", key);
+        }
+    }
 }

object_store/src/azure/builder.rs

@@ -408,7 +408,7 @@ impl FromStr for AzureConfigKey {
             "azure_disable_tagging" | "disable_tagging" => Ok(Self::DisableTagging),
             // Backwards compatibility
             "azure_allow_http" => Ok(Self::Client(ClientConfigKey::AllowHttp)),
-            _ => match s.parse() {
+            _ => match s.strip_prefix("azure_").unwrap_or(s).parse() {
                 Ok(key) => Ok(Self::Client(key)),
                 Err(_) => Err(Error::UnknownConfigurationKey { key: s.into() }.into()),
             },
@@ -1103,4 +1103,17 @@ mod tests {
         let pairs = split_sas(raw_sas).unwrap();
         assert_eq!(expected, pairs);
     }
+
+    #[test]
+    fn azure_test_client_opts() {
+        let key = "AZURE_PROXY_URL";
+        if let Ok(config_key) = key.to_ascii_lowercase().parse() {
+            assert_eq!(
+                AzureConfigKey::Client(ClientConfigKey::ProxyUrl),
+                config_key
+            );
+        } else {
+            panic!("{} not propagated as ClientConfigKey", key);
+        }
+    }
 }

object_store/src/client/mod.rs

@@ -157,6 +157,8 @@ impl FromStr for ClientConfigKey {
             "pool_idle_timeout" => Ok(Self::PoolIdleTimeout),
             "pool_max_idle_per_host" => Ok(Self::PoolMaxIdlePerHost),
             "proxy_url" => Ok(Self::ProxyUrl),
+            "proxy_ca_certificate" => Ok(Self::ProxyCaCertificate),
+            "proxy_excludes" => Ok(Self::ProxyExcludes),
             "timeout" => Ok(Self::Timeout),
             "user_agent" => Ok(Self::UserAgent),
             _ => Err(super::Error::UnknownConfigurationKey {

object_store/src/gcp/builder.rs

@@ -185,7 +185,7 @@ impl FromStr for GoogleConfigKey {
             "google_service_account_key" | "service_account_key" => Ok(Self::ServiceAccountKey),
             "google_bucket" | "google_bucket_name" | "bucket" | "bucket_name" => Ok(Self::Bucket),
             "google_application_credentials" => Ok(Self::ApplicationCredentials),
-            _ => match s.parse() {
+            _ => match s.strip_prefix("google_").unwrap_or(s).parse() {
                 Ok(key) => Ok(Self::Client(key)),
                 Err(_) => Err(Error::UnknownConfigurationKey { key: s.into() }.into()),
             },
@@ -671,4 +671,17 @@ mod tests {
             google_bucket_name
         );
     }
+
+    #[test]
+    fn gcp_test_client_opts() {
+        let key = "GOOGLE_PROXY_URL";
+        if let Ok(config_key) = key.to_ascii_lowercase().parse() {
+            assert_eq!(
+                GoogleConfigKey::Client(ClientConfigKey::ProxyUrl),
+                config_key
+            );
+        } else {
+            panic!("{} not propagated as ClientConfigKey", key);
+        }
+    }
 }