laskoviymishka commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4326559604
##########
crates/iceberg/src/io/file_io.rs:
##########
@@ -123,14 +146,38 @@ impl FileIO {
///
/// All storage configuration properties are included in the serialized
representation. These
/// properties may contain credentials or other sensitive values, so the
returned bytes must be
- /// protected in transit and at rest by the application embedding this
crate.
+ /// protected in transit and at rest by the application embedding this
crate. A serialized
+ /// credential provider may likewise carry catalog authentication and
vended credentials.
///
/// Storage factories are serialized through
[`typetag`](https://docs.rs/typetag). Third-party
/// factories must use `#[typetag::serde]` on their [`StorageFactory`]
implementation.
+ ///
+ /// A credential provider is serialized as the
+ /// [`StorageCredentialProviderFactory`] returned by
+ /// [`StorageCredentialProvider::factory`], and rebuilt on
deserialization. A provider that
+ /// cannot be rebuilt in another process is left out with a warning, as
with
+ /// [`FileIO::without_credential_provider`]: the deserialized `FileIO`
uses the credentials in
+ /// its configuration, which are not refreshed. Storage factories that
hold their own
+ /// credential sources, such as a custom AWS credential loader, still fail
to serialize, as
+ /// leaving them out would leave no credentials at all.
pub fn serialize_all(&self) -> Result<Vec<u8>> {
+ let credential_provider =
self.credential_provider.as_ref().and_then(|provider| {
+ provider
+ .factory()
+ .and_then(|factory| Ok(serde_json::to_value(factory)?))
+ .inspect_err(|error| {
+ static WARNED: Once = Once::new();
+ warn_once_without_provider(&WARNED, || {
+ format!("serializing FileIO without its credential
provider: {error}")
+ })
+ })
+ .ok()
Review Comment:
The default `factory()` errors, so any provider that doesn't opt in gets
stripped here. The failure then surfaces much later on a worker as an
`ExpiredToken`/403, far from this one-time warn. `OpenDalStorage` already
refuses to serialize in this situation, and `without_credential_provider()`
already exists as the explicit opt-out. So I'd return an error by default and
let callers who want best-effort ask for it.
The deserialize side is different: a worker that doesn't link the factory
can't do anything about it, so warn-and-drop seems right there. wdyt?
##########
crates/iceberg/src/io/storage/config/azdls.rs:
##########
@@ -36,6 +36,14 @@ pub const ADLS_ACCOUNT_NAME: &str = "adls.account-name";
pub const ADLS_ACCOUNT_KEY: &str = "adls.account-key";
/// The shared access signature.
pub const ADLS_SAS_TOKEN: &str = "adls.sas-token";
+/// Prefix for account-specific shared access signatures vended by a catalog.
+pub const ADLS_SAS_TOKEN_PREFIX: &str = "adls.sas-token.";
+/// Prefix for the epoch-millisecond expiration of an account-specific SAS
token.
+pub const ADLS_SAS_TOKEN_EXPIRES_AT_MS_PREFIX: &str =
"adls.sas-token-expires-at-ms.";
+/// Table property naming the endpoint a catalog serves to refresh vended ADLS
credentials.
+pub const ADLS_REFRESH_CREDENTIALS_ENDPOINT: &str =
"adls.refresh-credentials-endpoint";
+/// Table property enabling refresh of vended ADLS credentials. Defaults to
`true`.
+pub const ADLS_REFRESH_CREDENTIALS_ENABLED: &str =
"adls.refresh-credentials-enabled";
Review Comment:
Nothing in this PR reads the ADLS constants, or the
`*_REFRESH_CREDENTIALS_ENDPOINT`/`*_ENABLED` ones in `s3.rs` and `gcs.rs`. They
all still go into `public-api.txt`, and "Defaults to `true`" describes behavior
that doesn't exist yet. I'd move them to the PR that wires up REST refresh (and
the ADLS adapter), so the names and the default get checked against a real
consumer before semver locks them in.
##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,201 @@ pub trait StorageFactory: Debug + Send + Sync {
/// A `Result` containing an `Arc<dyn Storage>` on success, or an error
/// if the storage could not be created.
fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+ /// Build a new Storage instance, optionally supplying a credential
provider
+ /// that the backend can call to obtain and refresh short-lived
credentials.
+ ///
+ /// Backends that cannot use the provider ignore it and use the credentials
+ /// in `config`, as they would without one. The default does so and logs a
+ /// warning once per process, since the storage's credentials are then not
+ /// refreshed. Factories that wrap another factory should forward the
+ /// provider to it, and factories whose storage needs no credentials can
+ /// ignore it silently.
+ fn build_with_credential_provider(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+ ) -> Result<Arc<dyn Storage>> {
+ if credential_provider.is_some() {
+ static WARNED: Once = Once::new();
Review Comment:
A `static` inside a provided trait method is one item for the whole program,
not one per `Self`; the Rust reference calls this out for default impls. Once
the first factory type warns, every other one stays silent, and the message
keeps naming that first type. That's the wrapper-factory case @mbutrovich
raised. This runs once per `FileIO` behind the `OnceLock`, so I'd drop the
`Once` and warn every time. If that's too noisy, dedupe on
`type_name::<Self>()` instead.
##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,201 @@ pub trait StorageFactory: Debug + Send + Sync {
/// A `Result` containing an `Arc<dyn Storage>` on success, or an error
/// if the storage could not be created.
fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+ /// Build a new Storage instance, optionally supplying a credential
provider
+ /// that the backend can call to obtain and refresh short-lived
credentials.
+ ///
+ /// Backends that cannot use the provider ignore it and use the credentials
+ /// in `config`, as they would without one. The default does so and logs a
+ /// warning once per process, since the storage's credentials are then not
+ /// refreshed. Factories that wrap another factory should forward the
+ /// provider to it, and factories whose storage needs no credentials can
+ /// ignore it silently.
+ fn build_with_credential_provider(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+ ) -> Result<Arc<dyn Storage>> {
+ if credential_provider.is_some() {
+ static WARNED: Once = Once::new();
+ WARNED.call_once(|| {
+ tracing::warn!(
+ "{} ignores storage credential providers; its storage uses
the credentials \
+ in its configuration, which are not refreshed",
+ std::any::type_name::<Self>()
+ )
+ });
+ }
+ self.build(config)
+ }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Debug output
+///
+/// [`FileIO`](crate::io::FileIO) and storage implementations print the
+/// provider in their `Debug` output, so the provider's `Debug` implementation
+/// must not expose credentials or other secrets.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently.
+/// Implementations must cache internally and only re-fetch when the current
+/// credential is at or near expiry; otherwise every object-store request could
+/// trigger a call back to the catalog.
+///
+/// Backends may treat a credential as stale before it expires, and load
+/// another on every request until they get a fresher one. The OpenDAL S3 and
+/// GCS backends do so within two minutes of expiry, and reject a newly loaded
+/// credential that expires within ten seconds. Implementations should
+/// therefore refresh early enough that the credential they return has more
+/// than two minutes left.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {
+ /// Return whether this provider supplies credentials for `path`.
+ ///
+ /// Backends replace their own credential chain with the provider only for
+ /// paths it supports, so a provider must return `false` for the storage it
+ /// does not serve, such as other schemes behind a resolving storage.
+ fn supports_path(&self, path: &str) -> bool;
+
+ /// Load a fresh credential for the storage location identified by `path`.
+ ///
+ /// `path` is an absolute location: usually the file being accessed, such
+ /// as `s3://bucket/warehouse/db/table/data/file.parquet`, but for a bulk
+ /// delete the location shared by a batch, which may be a storage root
+ /// (`s3://bucket/`) or the prefix of a previously returned credential
+ /// (`s3://bucket/warehouse/db/table`). Providers that vend distinct
+ /// credentials per location prefix use it to select the most specific
+ /// match, which must [cover](StorageCredential::covers) `path`.
+ ///
+ /// Backends read the credential's expiry from its config, such as
+ /// `s3.session-token-expires-at-ms`, and load a new one before it expires.
+ /// A credential without an expiry is used for as long as the backend
lives.
+ ///
+ /// An error, or a credential that does not cover `path`, fails the
request:
+ /// backends do not fall back to other credentials.
+ async fn load_credential(&self, path: &str) -> Result<StorageCredential>;
Review Comment:
Since every provider has to pick the prefix itself, I'd ship the
longest-prefix rule as a helper in core and point these docs at it, something
like `StorageCredential::select<'a>(credentials: &'a [StorageCredential],
location: &str) -> Option<&'a StorageCredential>`. Java does this in one place
(`S3FileIO#clientForStoragePath` takes the longest matching prefix). Here
`covers` accepts any match, so a provider that does first-match over `[s3,
s3://bucket/table]` hands back the broader credential and nothing downstream
can tell.
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -696,17 +888,26 @@ impl Storage for OpenDalStorage {
}
async fn delete_stream(&self, mut paths: BoxStream<'static, String>) ->
Result<()> {
- let mut deleters: HashMap<String, opendal::Deleter> = HashMap::new();
+ let mut deleters: HashMap<DeleteBatchKey, opendal::Deleter> =
HashMap::new();
while let Some(path) = paths.next().await {
- let bucket = self.batch_key_for_path(&path);
+ let batch_key = self.delete_batch_key_for_path(&path).await?;
Review Comment:
Bulk delete now awaits `load_credential` once per path, where before it made
no provider calls. An orphan cleanup over 100k files is 100k provider calls,
and throughput depends entirely on the provider's cache. For `Scoped` batches
we could likely skip the call when an open batch's prefix already covers the
path. The scheme-wide case still needs the call to notice a scope change.
Either way, I'd add a test that counts provider calls for N paths so this can't
regress unnoticed.
##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,201 @@ pub trait StorageFactory: Debug + Send + Sync {
/// A `Result` containing an `Arc<dyn Storage>` on success, or an error
/// if the storage could not be created.
fn build(&self, config: &StorageConfig) -> Result<Arc<dyn Storage>>;
+
+ /// Build a new Storage instance, optionally supplying a credential
provider
+ /// that the backend can call to obtain and refresh short-lived
credentials.
+ ///
+ /// Backends that cannot use the provider ignore it and use the credentials
+ /// in `config`, as they would without one. The default does so and logs a
+ /// warning once per process, since the storage's credentials are then not
+ /// refreshed. Factories that wrap another factory should forward the
+ /// provider to it, and factories whose storage needs no credentials can
+ /// ignore it silently.
+ fn build_with_credential_provider(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+ ) -> Result<Arc<dyn Storage>> {
+ if credential_provider.is_some() {
+ static WARNED: Once = Once::new();
+ WARNED.call_once(|| {
+ tracing::warn!(
+ "{} ignores storage credential providers; its storage uses
the credentials \
+ in its configuration, which are not refreshed",
+ std::any::type_name::<Self>()
+ )
+ });
+ }
+ self.build(config)
+ }
+}
+
+/// Supplies fresh, backend-specific storage credentials on demand.
+///
+/// A catalog that vends temporary credentials implements this trait so that
+/// storage backends can re-fetch credentials as they approach expiry instead
+/// of failing once the initial token's TTL runs out.
+///
+/// # Debug output
+///
+/// [`FileIO`](crate::io::FileIO) and storage implementations print the
+/// provider in their `Debug` output, so the provider's `Debug` implementation
+/// must not expose credentials or other secrets.
+///
+/// # Caching
+///
+/// [`load_credential`](Self::load_credential) may be called very frequently.
+/// Implementations must cache internally and only re-fetch when the current
+/// credential is at or near expiry; otherwise every object-store request could
+/// trigger a call back to the catalog.
+///
+/// Backends may treat a credential as stale before it expires, and load
+/// another on every request until they get a fresher one. The OpenDAL S3 and
+/// GCS backends do so within two minutes of expiry, and reject a newly loaded
+/// credential that expires within ten seconds. Implementations should
+/// therefore refresh early enough that the credential they return has more
+/// than two minutes left.
+#[async_trait]
+pub trait StorageCredentialProvider: Debug + Send + Sync {
+ /// Return whether this provider supplies credentials for `path`.
+ ///
+ /// Backends replace their own credential chain with the provider only for
+ /// paths it supports, so a provider must return `false` for the storage it
+ /// does not serve, such as other schemes behind a resolving storage.
+ fn supports_path(&self, path: &str) -> bool;
+
+ /// Load a fresh credential for the storage location identified by `path`.
+ ///
+ /// `path` is an absolute location: usually the file being accessed, such
+ /// as `s3://bucket/warehouse/db/table/data/file.parquet`, but for a bulk
+ /// delete the location shared by a batch, which may be a storage root
+ /// (`s3://bucket/`) or the prefix of a previously returned credential
+ /// (`s3://bucket/warehouse/db/table`). Providers that vend distinct
+ /// credentials per location prefix use it to select the most specific
+ /// match, which must [cover](StorageCredential::covers) `path`.
+ ///
+ /// Backends read the credential's expiry from its config, such as
+ /// `s3.session-token-expires-at-ms`, and load a new one before it expires.
+ /// A credential without an expiry is used for as long as the backend
lives.
+ ///
+ /// An error, or a credential that does not cover `path`, fails the
request:
+ /// backends do not fall back to other credentials.
+ async fn load_credential(&self, path: &str) -> Result<StorageCredential>;
+
+ /// Return a factory that rebuilds an equivalent provider in another
process.
+ ///
+ /// [`FileIO::serialize_all`](crate::io::FileIO::serialize_all) serializes
this
+ /// factory in place of the provider. On an error, which the default always
+ /// returns, it serializes the `FileIO` without the provider and logs a
+ /// warning.
+ fn factory(&self) -> Result<Arc<dyn StorageCredentialProviderFactory>> {
+ Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ "storage credential provider cannot be serialized",
+ ))
+ }
+}
+
+/// Serializable recipe that rebuilds a [`StorageCredentialProvider`] after
+/// [`FileIO`](crate::io::FileIO) deserialization.
+///
+/// Factories are serialized through [`typetag`](https://docs.rs/typetag), so
+/// implementations must use `#[typetag::serde]`, and the receiving binary must
+/// link the concrete implementation.
+#[typetag::serde(tag = "type")]
+pub trait StorageCredentialProviderFactory: Debug + Send + Sync {
+ /// Build a provider for a `FileIO` with the given storage configuration.
+ fn build(&self, config: &StorageConfig) -> Result<Arc<dyn
StorageCredentialProvider>>;
+}
+
+/// A vended storage credential: the storage properties that apply to
+/// locations under a prefix, like Java's `StorageCredential`.
+///
+/// `config` holds backend storage properties, such as `s3.access-key-id`,
+/// `s3.secret-access-key`, `s3.session-token` and
+/// `s3.session-token-expires-at-ms`. Backends read their credential and its
+/// expiry from it.
+#[derive(Clone, PartialEq, Eq)]
+pub struct StorageCredential {
+ /// Storage-location prefix this credential applies to, such as
+ /// `s3://bucket/table` or, for a whole scheme, `s3`.
+ prefix: String,
+ /// Backend storage properties holding the credential.
+ config: HashMap<String, String>,
+}
+
+impl StorageCredential {
+ /// Create a credential for locations under `prefix`.
+ ///
+ /// An empty prefix covers no location; Java rejects one outright.
+ pub fn new(prefix: impl Into<String>, config: HashMap<String, String>) ->
Self {
+ Self {
+ prefix: prefix.into(),
+ config,
+ }
+ }
+
+ /// Return the storage-location prefix this credential applies to.
+ pub fn prefix(&self) -> &str {
+ &self.prefix
+ }
+
+ /// Return the storage properties holding the credential.
+ pub fn config(&self) -> &HashMap<String, String> {
+ &self.config
+ }
+
+ /// Return whether this credential applies to `location`.
+ ///
+ /// As in Java, `location` must start with the non-empty prefix, compared
+ /// as plain strings: `s3` covers every `s3://`, `s3a://` and `s3n://`
+ /// location, and `s3://bucket/tab` covers `s3://bucket/table`.
+ pub fn covers(&self, location: &str) -> bool {
+ !self.prefix.is_empty() && location.starts_with(&self.prefix)
+ }
+}
+
+impl Debug for StorageCredential {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ // Config values are secrets.
+ f.debug_struct("StorageCredential")
+ .field("prefix", &self.prefix)
+ .field("config_keys", &self.config.keys().collect::<Vec<_>>())
Review Comment:
small thing: `HashMap` key order is randomized per process, so this `Debug`
output changes from run to run. I'd sort the keys before printing so logs and
any snapshot tests stay stable.
##########
crates/storage/opendal/src/s3.rs:
##########
@@ -148,11 +169,49 @@ pub(crate) fn s3_config_build(
if let Some(loader) = customized_credential_load {
let chain = ProvideCredentialChain::new().push(Arc::clone(&loader.0));
builder = builder.credential_provider_chain(chain);
+ } else if let Some(provider) = credential_provider {
Review Comment:
GCS clears every competing source in `disable_other_credential_sources`.
Here we only install the chain and leave
`access_key_id`/`secret_access_key`/`session_token`, `role_arn` and config/IMDS
loading in `cfg`. The comment above promises that refresh failures can't fall
through. That only holds if OpenDAL's S3 builder ignores those fields once a
custom chain is set, and doesn't, say, assume `role_arn` on top of the vended
credentials. Nothing tests that. The S3 tests set
`disable_config_load`/`disable_ec2_metadata` and no static keys, so they can't
catch a fallback.
I'd add the S3 twin of
`test_gcs_does_not_fall_back_when_the_provider_fails`: static keys and a role
ARN in config, plus a failing provider. If that ever signs a request, clear
those fields the way GCS does. REST catalogs often leave `s3.access-key-id` in
the props next to vended credentials, so this is the realistic setup.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
