You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "zakariya-s (via GitHub)" <gi...@apache.org> on 2026/08/07 10:31:35 UTC
[PR] [1/6] feat: add refreshable storage credential providers [iceberg-rust]
zakariya-s opened a new pull request, #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
- Part of #2931.
- Extracted from #2932.
## What changes are included in this PR?
<!--
Provide a summary of the modifications in this PR. List the main changes such as new features, bug fixes, refactoring, or any other updates.
-->
This PR introduces the core API for supplying refreshable storage credentials:
- Adds `StorageCredentialProvider` and backend-specific credential types for S3 and GCS.
- Propagates an optional credential provider through `FileIOBuilder` and `FileIO`.
- Adds `StorageFactory::build_with_credentials`.
- Adds the standard S3 and GCS credential refresh property constants.
Backend integration and REST credential refreshing will be added in subsequent PRs.
## Are these changes tested?
<!--
Specify what test covers (unit test, integration test, etc.).
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
Yes.
## AI Disclosure
<!--
https://iceberg.apache.org/contribute/#guidelines-for-ai-assisted-contributions
-->
AI helped splitting this from the reference PR.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/6] feat: add refreshable storage credential providers [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-5681346083
not stale
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4376494625
##########
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:
added `StorageCredential::select` and pointed the `load_credential` docs at it
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4204970918
##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,179 @@ 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 exactly that.
+ #[allow(unused_variables)]
+ fn build_with_credential_provider(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+ ) -> Result<Arc<dyn Storage>> {
+ self.build(config)
+ }
Review Comment:
done in 7ff3d887594502ca2dd04d4b7b95ce3b8c46c6f6, but i've also overridden the in-memory and local FS factories to ignore the provider without a warning log msg since it's not relevant for them
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4206008061
##########
crates/storage/opendal/Cargo.toml:
##########
@@ -57,14 +57,17 @@ iceberg = { workspace = true }
iceberg-property-macro = { workspace = true }
opendal = { workspace = true }
reqsign-aws-v4 = { version = "3.0.0", optional = true }
-reqsign-core = { version = "3.0.0", optional = true }
+reqsign-core = { version = "3.3.1", optional = true }
Review Comment:
reqsign-core 3.3.1 deduplicates concurrent credential refreshes (apache/reqsign#842), so concurrent requests on one operator call the provider once per refresh
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/N] feat(io): support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4200705516
##########
Cargo.lock:
##########
@@ -6193,10 +6212,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
dependencies = [
"fastrand",
- "getrandom 0.4.3",
+ "getrandom 0.3.4",
"once_cell",
"rustix",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
Review Comment:
oops good catch, fixed
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/N] feat(io): support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-6026150222
Thanks for the review @mbutrovich! I've addressed your comments and have updated the task list in #2931. I decided to combine both the core API and the S3/GCS backends so that it's clear how the API would be used. I'll create a follow-up PR eventually for those independent fixes.
If we're okay with this PR, I would probably try to follow this PR with the ADLS support, as per the updated task list. At least then we can also bump OpenDAL to 0.59 too.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-6415019229
thanks @laskoviymishka for the thorough review! i think i've addressed all the comments, would really appreciate another pass when you have time! i'll also look at the comments in the reference PR too
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4210829243
##########
crates/iceberg/src/io/storage/local_fs.rs:
##########
@@ -318,6 +318,16 @@ impl StorageFactory for LocalFsStorageFactory {
fn build(&self, _config: &StorageConfig) -> Result<Arc<dyn Storage>> {
Ok(Arc::new(LocalFsStorage::new()))
}
+
+ /// Local filesystem storage needs no credentials.
+ #[allow(unused_variables)]
+ fn build_with_credential_provider(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
Review Comment:
8455a06dd4e25b74861d9c953546dbcf28da6d6e
##########
crates/iceberg/src/io/storage/memory.rs:
##########
@@ -242,6 +242,16 @@ impl StorageFactory for MemoryStorageFactory {
fn build(&self, _config: &StorageConfig) -> Result<Arc<dyn Storage>> {
Ok(Arc::new(MemoryStorage::new()))
}
+
+ /// In-memory storage needs no credentials.
+ #[allow(unused_variables)]
+ fn build_with_credential_provider(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
Review Comment:
8455a06dd4e25b74861d9c953546dbcf28da6d6e
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-6120736264
Hey @blackmwk @CTTY I'd love if there could be some eyes on this PR whenever you have time!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4376462713
##########
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:
added `test_s3_does_not_fall_back_when_the_provider_fails`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/6] feat: add refreshable storage credential providers [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-5272671066
Hey @paleolimbot, great to hear this will be useful! We also have to do workarounds at the moment to get this working for our DataFusion clients, and I've also heard that others are doing similar stuff in their internal forks of iceberg-rust.
Like you mention, this PR doesn't quite implement the entire E2E refreshing, but I do plan to split it into a few PRs to ease the review burden. Feel free to take a look at the full draft PR (https://github.com/apache/iceberg-rust/pull/2932) which does implement it E2E and gives an indicator of how it should roughly look like when done.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/6] feat: add refreshable storage credential providers [iceberg-rust]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-5672777994
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/N] feat(io): support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-6020780575
@CTTY would it be possible to get a review too please?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/N] feat(io): support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4200715314
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -278,6 +306,7 @@ fn default_memory_operator() -> Operator {
///
/// The serialized representation is not a stable format and may change between crate versions.
#[derive(Clone, Debug, Serialize, Deserialize)]
+#[non_exhaustive]
Review Comment:
[done here](https://github.com/apache/iceberg-rust/pull/2976/changes/e5e50f39960c320072b1dd852748d668ac766b11#diff-06572a96a58dc510037d5efa622f9bec8519bc1beab13c9f251e97e657a9d4edR36)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/N] feat(io): support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-6020682368
@mbutrovich thanks for reviewing the reference PR! I've split things here, without ADLS support or the main REST catalog provider.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/N] feat(io): support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4198816056
##########
Cargo.lock:
##########
@@ -6193,10 +6212,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
dependencies = [
"fastrand",
- "getrandom 0.4.3",
+ "getrandom 0.3.4",
"once_cell",
"rustix",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
Review Comment:
The lockfile also moves several crates to older versions that the new dependencies don't need. `windows-sys` goes from 0.61.2 to 0.48.0 or 0.52.0 for `errno`, `rustix`, `tempfile` and others, and `getrandom` goes from 0.4.3 to 0.3.4 for `tempfile`. When I start from `main`'s `Cargo.lock` with this PR's manifests and run `cargo metadata`, the only version changes are `reqsign-aws-core`, `reqsign-aws-v4`, `reqsign-core`, a second `quick-xml`, `asyncband`, and `socket2`. Could you regenerate the lockfile that way, so the PR doesn't carry the unrelated downgrades?
##########
crates/storage/opendal/src/gcs.rs:
##########
@@ -45,7 +51,9 @@ pub(crate) fn gcs_config_parse(mut m: HashMap<String, String>) -> Result<GcsConf
cfg.endpoint = Some(endpoint);
}
- if m.remove(GCS_NO_AUTH).is_some() {
+ if let Some(no_auth) = m.remove(GCS_NO_AUTH)
+ && is_truthy(&no_auth)
+ {
Review Comment:
Could the `gcs.no-auth` parsing fix move to its own PR, along with the other fixes that don't depend on credential refresh? I count three of them in this PR:
- `gcs.no-auth` now enables anonymous access only for a truthy value. On `main`, any value, including `false`, turns it on.
- GCS now accepts `gcs://` paths in [`create_operator`](https://github.com/apache/iceberg-rust/blob/f631771198df4b98b251c391b43aadde060579cb/crates/storage/opendal/src/lib.rs#L557-L563), [`relativize_path`](https://github.com/apache/iceberg-rust/blob/f631771198df4b98b251c391b43aadde060579cb/crates/storage/opendal/src/lib.rs#L760-L772), and [`gcs_config_build`](https://github.com/apache/iceberg-rust/blob/f631771198df4b98b251c391b43aadde060579cb/crates/storage/opendal/src/gcs.rs#L89-L94). On `main`, `OpenDalResolvingStorage` routes `gcs://` to the GCS backend, but the backend then rejects the path.
- [`OpenDalResolvingStorage`'s `Debug`](https://github.com/apache/iceberg-rust/blob/f631771198df4b98b251c391b43aadde060579cb/crates/storage/opendal/src/resolving.rs#L263-L270) now prints property keys only. On `main` it prints the property values, including secrets.
Each one changes behavior for users who never set a credential provider, so it deserves its own changelog entry, and a reviewer can check it without reading the refresh code. They line up with the first piece of the split discussed on #2932.
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -278,6 +306,7 @@ fn default_memory_operator() -> Operator {
///
/// The serialized representation is not a stable format and may change between crate versions.
#[derive(Clone, Debug, Serialize, Deserialize)]
+#[non_exhaustive]
Review Comment:
Could you add an entry for this change under `[Unreleased]` "Breaking Changes" in [`CHANGELOG.md`](https://github.com/apache/iceberg-rust/blob/ecba0b537a1f78ec8125c1a26f6dfab3f95b509e/CHANGELOG.md?plain=1#L27-L35)? Adding `#[non_exhaustive]` to `OpenDalStorage` and its variants, plus the new `credential_provider` fields on `S3` and `Gcs`, breaks downstream code that builds a variant with a struct literal or matches the enum exhaustively (`crates/storage/opendal/public-api.txt` shows the change). An entry tells those users what to change, the way #3286 did for `BoxedCatalogBuilder::with_runtime`. We discussed flagging this break in the split PR that carries it on #2932 ([thread](https://github.com/apache/iceberg-rust/pull/2932#discussion_r4137009802)).
##########
crates/storage/opendal/src/utils.rs:
##########
@@ -27,3 +28,296 @@ pub(crate) fn from_opendal_error(e: opendal::Error) -> iceberg::Error {
)
.with_source(e)
}
+
+/// The non-empty value of `key` in a vended credential's config.
+#[cfg(any(feature = "opendal-s3", feature = "opendal-gcs"))]
+pub(crate) fn required_credential_property<'a>(
+ config: &'a std::collections::HashMap<String, String>,
Review Comment:
The new code in `utils.rs` spells out `std::collections::HashMap`, `std::sync::Arc`, `iceberg::io::StorageCredential`, `iceberg::io::StorageCredentialProvider`, and `reqsign_core::...` inline throughout. If I'm reading it right, that's because `reqsign_core` is only available with the `opendal-s3` or `opendal-gcs` feature. Could these be `use` items gated with `#[cfg(any(feature = "opendal-s3", feature = "opendal-gcs"))]`, like the rest of the crate imports its names? The same goes for the three `std::sync::Once` uses in [`file_io.rs`](https://github.com/apache/iceberg-rust/blob/f631771198df4b98b251c391b43aadde060579cb/crates/iceberg/src/io/file_io.rs#L104), where `Once` can join the existing `use std::sync::{Arc, OnceLock};`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4204978950
##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,179 @@ 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 exactly that.
+ #[allow(unused_variables)]
+ fn build_with_credential_provider(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+ ) -> Result<Arc<dyn Storage>> {
+ 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.
Review Comment:
thanks, added a mention of reqsign's 2 min reload window in 7ff3d887594502ca2dd04d4b7b95ce3b8c46c6f6
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/6] feat: add refreshable storage credential providers [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-5300999093
Thanks @CTTY, the main reason for the trait is that credential acquisition/vending isn't really backend specific, and doing it this way avoids each backend storage factory reimplementing the REST auth logic (e.g. auth, parsing, prefix selection, caching, backoff, etc.). Without the trait, `RestCatalog` would have to know or downcast each StorageFactory to configure each specific credential loader.
This approach also opens the possibility of non-REST implementations too (like Glue/Lake Formation via AWS specific APIs) without doing what Java does with having custom credential providers for each backend.
Also Iceberg Java has the backend-specific vended credential providers in each of the cloud-specific `FileIO` implementations, whereas we only have a generic `FileIO` in iceberg-rust as OpenDAL handles most of it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/N] feat(io): support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-6021924412
> @CTTY would it be possible to get a review too please?
I will take a look today too.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "laskoviymishka (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4376488642
##########
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:
oops good catch, each factory type now warns once. i thought that warning on every build would be noisy for engines that deserialise a FileIO per task, so i think deduping is better than warning every time.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/6] feat: add refreshable storage credential providers [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-5293086272
Hi @CTTY! I know that bandwidth has been tough at the moment, but could I get a first pass at this (and the reference https://github.com/apache/iceberg-rust/pull/2932 for the bigger context) when you have time (no rush at all for this PR). Thank you!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] [1/N] feat(io): support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4200709768
##########
crates/storage/opendal/src/gcs.rs:
##########
@@ -45,7 +51,9 @@ pub(crate) fn gcs_config_parse(mut m: HashMap<String, String>) -> Result<GcsConf
cfg.endpoint = Some(endpoint);
}
- if m.remove(GCS_NO_AUTH).is_some() {
+ if let Some(no_auth) = m.remove(GCS_NO_AUTH)
+ && is_truthy(&no_auth)
+ {
Review Comment:
yes i'll try to add it in a follow-up PR, thanks
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4376588190
##########
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:
added `test_delete_stream_loads_a_credential_per_path_and_batch`, i'm not sure whether we should skip the call as `S3FileIO.deleteFiles` also does [an in-memory longest prefix lookup for every path](https://github.com/apache/iceberg/blob/47677abf352f2e5c373cf82704ea7f2e675f8862/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java#L167). since we require providers to cache, this is effectively the same.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4376716142
##########
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:
i'm okay with either behaviour, @mbutrovich suggested to warn in the reference pr: https://github.com/apache/iceberg-rust/pull/2932#discussion_r4136994726. i do agree that this will eventually fail for a long task once the token expires, but unlike java we don't have reflection so we can't load AuthManager dynamically. we could maybe make them rebuildable if we use typetag like we have done so for providers? that should probably be in a follow-up PR though. for now should i make this fail?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4201140750
##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,179 @@ 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 exactly that.
+ #[allow(unused_variables)]
+ fn build_with_credential_provider(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+ ) -> Result<Arc<dyn Storage>> {
+ self.build(config)
+ }
Review Comment:
`FileIO` builds every storage through this method, so a factory that implements only `build` drops the provider and logs nothing. A `RestCatalog` accepts a user's factory through [`CatalogBuilder::with_storage_factory`](https://github.com/apache/iceberg-rust/blob/f2409e9f5eaff1c643985013d927e17fdf16e6ec/crates/iceberg/src/catalog/mod.rs#L157), for example a wrapper around `OpenDalStorageFactory` that adds metrics. As I read it, once the catalog wiring lands, a wrapper like that would keep using the initial vended credentials until they expire, and the first symptom would be an authorization error from the object store. Could the default log a warning once when it drops a provider, the way `serialize_all` does with [`warn_once_without_provider`](https://github.com/apache/iceberg-rust/blob/e5e50f39960c320072b1dd852748d668ac766b11/crates/iceberg/src/io/file_io.rs#L101-L111)?
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -1098,11 +1354,353 @@ mod tests {
}
}
+ /// Vends a credential for the whole `s3` scheme until it has served
+ /// `table-c`, then credentials scoped to each table. The scheme-wide
+ /// credential expires once the `table-c` batch is signed.
+ #[cfg(feature = "opendal-s3")]
+ #[derive(Debug, Default)]
+ struct ChangingScopeProvider {
+ scoped: AtomicBool,
+ scheme_wide_expired: AtomicBool,
+ }
+
+ #[cfg(feature = "opendal-s3")]
+ #[async_trait]
+ impl StorageCredentialProvider for ChangingScopeProvider {
+ fn supports_path(&self, _path: &str) -> bool {
+ true
+ }
+
+ async fn load_credential(&self, path: &str) -> Result<StorageCredential> {
+ if path.starts_with("s3://bucket/table-c") {
+ self.scoped.store(true, Ordering::SeqCst);
+ }
+ // Signing the scoped batch means the scheme-wide credential is gone.
+ if path == "s3://bucket/table-c" {
+ self.scheme_wide_expired.store(true, Ordering::SeqCst);
+ }
+ match path.strip_prefix("s3://bucket/") {
+ Some(rest) if self.scoped.load(Ordering::SeqCst) && !rest.is_empty() => {
+ let table = rest.split('/').next().unwrap_or_default();
+ Ok(s3_credential(format!("s3://bucket/{table}"), "SCOPED_AK"))
+ }
+ _ if self.scheme_wide_expired.load(Ordering::SeqCst) => Err(Error::new(
+ ErrorKind::Unexpected,
+ "the scheme-wide credential expired",
+ )),
+ _ => Ok(s3_credential("s3", "SCHEME_AK")),
+ }
+ }
+ }
+
+ #[cfg(feature = "opendal-s3")]
+ #[tokio::test]
+ async fn test_delete_stream_signs_each_batch_with_its_vended_credential() {
+ let mut server = mockito::Server::new_async().await;
+ let mut delete = |path: &str, access_key: &str| {
+ server
+ .mock("DELETE", path)
+ .match_header(
+ "authorization",
+ mockito::Matcher::Regex(format!("Credential={access_key}/")),
+ )
+ .expect(1)
+ .with_status(204)
+ };
+ // The scheme-wide batch is flushed, with its own credential, before
+ // the scoped credential that replaces it is used.
+ let scheme_wide = delete("/bucket/table-b/f.parquet", "SCHEME_AK")
+ .create_async()
+ .await;
+ let scoped = delete("/bucket/table-c/f.parquet", "SCOPED_AK")
+ .create_async()
+ .await;
+
+ let mut config = S3Config::default();
+ config.endpoint = Some(server.url());
+ config.region = Some("us-east-1".to_string());
+ config.disable_config_load = true;
+ config.disable_ec2_metadata = true;
+ let storage = OpenDalStorage::S3 {
+ config: Arc::new(config),
+ customized_credential_load: None,
+ credential_provider: Some(Arc::new(ChangingScopeProvider::default())),
+ client_config: OpenDalClientConfig::default(),
+ };
+ storage
+ .delete_stream(
+ futures::stream::iter([
+ "s3://bucket/table-b/f.parquet".to_string(),
+ "s3://bucket/table-c/f.parquet".to_string(),
+ ])
+ .boxed(),
+ )
+ .await
+ .unwrap();
+
+ scheme_wide.assert_async().await;
+ scoped.assert_async().await;
+ }
+
+ /// Vends `token` as a GCS credential, or fails when it is `None`.
+ #[cfg(feature = "opendal-gcs")]
+ #[derive(Debug)]
+ struct GcsTokenProvider(Option<&'static str>);
+
+ #[cfg(feature = "opendal-gcs")]
+ #[async_trait]
+ impl StorageCredentialProvider for GcsTokenProvider {
+ fn supports_path(&self, _path: &str) -> bool {
+ true
+ }
+
+ async fn load_credential(&self, _path: &str) -> Result<StorageCredential> {
+ let token = self
+ .0
+ .ok_or_else(|| Error::new(ErrorKind::Unexpected, "refresh failed"))?;
+ Ok(StorageCredential::new(
+ "gs",
+ HashMap::from([(GCS_TOKEN.to_string(), token.to_string())]),
+ ))
+ }
+ }
+
+ #[cfg(feature = "opendal-gcs")]
+ fn gcs_storage(server: &mockito::Server, provider: GcsTokenProvider) -> OpenDalStorage {
+ let config = gcs_config_parse(HashMap::from([
+ (GCS_SERVICE_HOST.to_string(), server.url()),
+ (GCS_TOKEN.to_string(), "static-token".to_string()),
+ ]))
+ .unwrap();
+ OpenDalStorage::Gcs {
+ config: Arc::new(config),
+ credential_provider: Some(Arc::new(provider)),
+ client_config: OpenDalClientConfig::default(),
+ }
+ }
+
+ #[cfg(feature = "opendal-gcs")]
+ #[tokio::test]
+ async fn test_gcs_signs_with_the_vended_token() {
+ let mut server = mockito::Server::new_async().await;
+ let mock = server
+ .mock("GET", mockito::Matcher::Any)
+ .match_header("authorization", "Bearer vended-token")
+ .expect_at_least(1)
+ .with_status(404)
+ .create_async()
+ .await;
+
+ let storage = gcs_storage(&server, GcsTokenProvider(Some("vended-token")));
+ assert!(!storage.exists("gs://bucket/file.parquet").await.unwrap());
+ mock.assert_async().await;
+ }
Review Comment:
Could you add a test where one operator signs a request, the credential nears expiry, and the next request goes out with a newly loaded credential? The tests here check the first signed request, and the adapter tests in `s3.rs` and `gcs.rs` check that the expiry is parsed. None of them covers the reload itself, which is the behavior this PR adds. For example, a provider could return a different token on each call with an expiry 60 seconds out, and the test could call `exists` twice on one operator from `create_operator`, with mocks that expect the first token and then the second. reqsign (the request-signing library OpenDAL uses) reloads inside its 120-second window, so the two requests carry different tokens. #2931 plans an OpenDAL 0.59 bump next, and a test like this would catch any change in how the newer reqsign caches credentials.
##########
crates/iceberg/src/io/storage/mod.rs:
##########
@@ -139,4 +140,179 @@ 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 exactly that.
+ #[allow(unused_variables)]
+ fn build_with_credential_provider(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
+ ) -> Result<Arc<dyn Storage>> {
+ 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.
Review Comment:
How close to expiry counts as "near" here? The S3 and GCS backends pass the credential to reqsign, the request-signing library OpenDAL uses. reqsign treats a cached credential as stale 120 seconds before it expires ([AWS](https://docs.rs/reqsign-aws-core/3.2.0/src/reqsign_aws_core/credential.rs.html#49-51), [GCS](https://docs.rs/reqsign-google/3.0.4/src/reqsign_google/credential.rs.html#248-250)) and calls the provider again on every request after that. It also rejects a reloaded credential that expires within 10 seconds ([`validate_refreshed_credential`](https://docs.rs/reqsign-core/3.3.2/src/reqsign_core/signer.rs.html#159-176), with the 10-second headroom for [AWS](https://docs.rs/reqsign-aws-v4/3.3.1/src/reqsign_aws_v4/sign_request.rs.html#32) and [GCS](https://docs.rs/reqsign-google/3.0.4/src/reqsign_google/sign_request.rs.html#35)).
I checked this at the head commit. I built one GCS operator with `create_operator`, used a provider that counts its calls and sets `gcs.oauth2.token-expires-at`, and called `exists` three times on that operator:
- With a token that expires in an hour, the provider was called once.
- With a token that expires in 60 seconds, the provider was called three times.
- With a token that expires in 5 seconds, every request failed with `refreshed signing credential expires before the requested operation deadline`.
Could this section tell implementers to return a credential with more than two minutes left, so they refresh earlier than that? A provider that refreshes one minute before expiry gets a call on every request for a minute. A provider that refreshes at expiry fails every request in the last 10 seconds.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4210146424
##########
crates/iceberg/src/io/storage/local_fs.rs:
##########
@@ -318,6 +318,16 @@ impl StorageFactory for LocalFsStorageFactory {
fn build(&self, _config: &StorageConfig) -> Result<Arc<dyn Storage>> {
Ok(Arc::new(LocalFsStorage::new()))
}
+
+ /// Local filesystem storage needs no credentials.
+ #[allow(unused_variables)]
+ fn build_with_credential_provider(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
Review Comment:
Could this name the parameter `_credential_provider` instead of using `#[allow(unused_variables)]`? `build` just above already marks its unused parameter as `_config`. The attribute also covers the whole function, so it would hide any unused variable added here later. `crates/iceberg/public-api.txt` would pick up the new parameter name.
```suggestion
fn build_with_credential_provider(
&self,
config: &StorageConfig,
_credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
```
##########
crates/iceberg/src/io/storage/memory.rs:
##########
@@ -242,6 +242,16 @@ impl StorageFactory for MemoryStorageFactory {
fn build(&self, _config: &StorageConfig) -> Result<Arc<dyn Storage>> {
Ok(Arc::new(MemoryStorage::new()))
}
+
+ /// In-memory storage needs no credentials.
+ #[allow(unused_variables)]
+ fn build_with_credential_provider(
+ &self,
+ config: &StorageConfig,
+ credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
Review Comment:
Could this also use `_credential_provider` in place of `#[allow(unused_variables)]`, to match `build(&self, _config: ...)` above?
```suggestion
fn build_with_credential_provider(
&self,
config: &StorageConfig,
_credential_provider: Option<Arc<dyn StorageCredentialProvider>>,
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-6045304477
Done thanks @mbutrovich! Please let me know if there's anything else to change
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#issuecomment-6041794628
Thanks @mbutrovich for the quick second review! I've made the appropriate changes. Would appreciate another round when you have time!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat(io)!: [1/N] support refreshable storage credentials for S3 and GCS [iceberg-rust]
Posted by "zakariya-s (via GitHub)" <gi...@apache.org>.
zakariya-s commented on code in PR #2976:
URL: https://github.com/apache/iceberg-rust/pull/2976#discussion_r4205003272
##########
crates/storage/opendal/src/lib.rs:
##########
@@ -1098,11 +1354,353 @@ mod tests {
}
}
+ /// Vends a credential for the whole `s3` scheme until it has served
+ /// `table-c`, then credentials scoped to each table. The scheme-wide
+ /// credential expires once the `table-c` batch is signed.
+ #[cfg(feature = "opendal-s3")]
+ #[derive(Debug, Default)]
+ struct ChangingScopeProvider {
+ scoped: AtomicBool,
+ scheme_wide_expired: AtomicBool,
+ }
+
+ #[cfg(feature = "opendal-s3")]
+ #[async_trait]
+ impl StorageCredentialProvider for ChangingScopeProvider {
+ fn supports_path(&self, _path: &str) -> bool {
+ true
+ }
+
+ async fn load_credential(&self, path: &str) -> Result<StorageCredential> {
+ if path.starts_with("s3://bucket/table-c") {
+ self.scoped.store(true, Ordering::SeqCst);
+ }
+ // Signing the scoped batch means the scheme-wide credential is gone.
+ if path == "s3://bucket/table-c" {
+ self.scheme_wide_expired.store(true, Ordering::SeqCst);
+ }
+ match path.strip_prefix("s3://bucket/") {
+ Some(rest) if self.scoped.load(Ordering::SeqCst) && !rest.is_empty() => {
+ let table = rest.split('/').next().unwrap_or_default();
+ Ok(s3_credential(format!("s3://bucket/{table}"), "SCOPED_AK"))
+ }
+ _ if self.scheme_wide_expired.load(Ordering::SeqCst) => Err(Error::new(
+ ErrorKind::Unexpected,
+ "the scheme-wide credential expired",
+ )),
+ _ => Ok(s3_credential("s3", "SCHEME_AK")),
+ }
+ }
+ }
+
+ #[cfg(feature = "opendal-s3")]
+ #[tokio::test]
+ async fn test_delete_stream_signs_each_batch_with_its_vended_credential() {
+ let mut server = mockito::Server::new_async().await;
+ let mut delete = |path: &str, access_key: &str| {
+ server
+ .mock("DELETE", path)
+ .match_header(
+ "authorization",
+ mockito::Matcher::Regex(format!("Credential={access_key}/")),
+ )
+ .expect(1)
+ .with_status(204)
+ };
+ // The scheme-wide batch is flushed, with its own credential, before
+ // the scoped credential that replaces it is used.
+ let scheme_wide = delete("/bucket/table-b/f.parquet", "SCHEME_AK")
+ .create_async()
+ .await;
+ let scoped = delete("/bucket/table-c/f.parquet", "SCOPED_AK")
+ .create_async()
+ .await;
+
+ let mut config = S3Config::default();
+ config.endpoint = Some(server.url());
+ config.region = Some("us-east-1".to_string());
+ config.disable_config_load = true;
+ config.disable_ec2_metadata = true;
+ let storage = OpenDalStorage::S3 {
+ config: Arc::new(config),
+ customized_credential_load: None,
+ credential_provider: Some(Arc::new(ChangingScopeProvider::default())),
+ client_config: OpenDalClientConfig::default(),
+ };
+ storage
+ .delete_stream(
+ futures::stream::iter([
+ "s3://bucket/table-b/f.parquet".to_string(),
+ "s3://bucket/table-c/f.parquet".to_string(),
+ ])
+ .boxed(),
+ )
+ .await
+ .unwrap();
+
+ scheme_wide.assert_async().await;
+ scoped.assert_async().await;
+ }
+
+ /// Vends `token` as a GCS credential, or fails when it is `None`.
+ #[cfg(feature = "opendal-gcs")]
+ #[derive(Debug)]
+ struct GcsTokenProvider(Option<&'static str>);
+
+ #[cfg(feature = "opendal-gcs")]
+ #[async_trait]
+ impl StorageCredentialProvider for GcsTokenProvider {
+ fn supports_path(&self, _path: &str) -> bool {
+ true
+ }
+
+ async fn load_credential(&self, _path: &str) -> Result<StorageCredential> {
+ let token = self
+ .0
+ .ok_or_else(|| Error::new(ErrorKind::Unexpected, "refresh failed"))?;
+ Ok(StorageCredential::new(
+ "gs",
+ HashMap::from([(GCS_TOKEN.to_string(), token.to_string())]),
+ ))
+ }
+ }
+
+ #[cfg(feature = "opendal-gcs")]
+ fn gcs_storage(server: &mockito::Server, provider: GcsTokenProvider) -> OpenDalStorage {
+ let config = gcs_config_parse(HashMap::from([
+ (GCS_SERVICE_HOST.to_string(), server.url()),
+ (GCS_TOKEN.to_string(), "static-token".to_string()),
+ ]))
+ .unwrap();
+ OpenDalStorage::Gcs {
+ config: Arc::new(config),
+ credential_provider: Some(Arc::new(provider)),
+ client_config: OpenDalClientConfig::default(),
+ }
+ }
+
+ #[cfg(feature = "opendal-gcs")]
+ #[tokio::test]
+ async fn test_gcs_signs_with_the_vended_token() {
+ let mut server = mockito::Server::new_async().await;
+ let mock = server
+ .mock("GET", mockito::Matcher::Any)
+ .match_header("authorization", "Bearer vended-token")
+ .expect_at_least(1)
+ .with_status(404)
+ .create_async()
+ .await;
+
+ let storage = gcs_storage(&server, GcsTokenProvider(Some("vended-token")));
+ assert!(!storage.exists("gs://bucket/file.parquet").await.unwrap());
+ mock.assert_async().await;
+ }
Review Comment:
yes good catch, made both an s3 [test_s3_operator_reloads_an_expiring_credential](https://github.com/apache/iceberg-rust/pull/2976/changes/7ff3d887594502ca2dd04d4b7b95ce3b8c46c6f6#diff-3c899976a8c7dc30d93ea1fea29403afdf76b1efcf175770cdfef47de4a22990R1511-R1548) and a gcs one [test_gcs_operator_reloads_an_expiring_token](https://github.com/apache/iceberg-rust/pull/2976/changes/7ff3d887594502ca2dd04d4b7b95ce3b8c46c6f6#diff-3c899976a8c7dc30d93ea1fea29403afdf76b1efcf175770cdfef47de4a22990R1639-R1662)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org