feat(rest): Support refreshing vended storage credentials - #2932
feat(rest): Support refreshing vended storage credentials#2932zakariya-s wants to merge 4 commits into
Conversation
| /// Disable header redaction in error logs (defaults to false for security) | ||
| pub const REST_CATALOG_PROP_DISABLE_HEADER_REDACTION: &str = "disable-header-redaction"; | ||
| /// Identifier for a server-side scan plan associated with credential requests. | ||
| pub const REST_CATALOG_PROP_SCAN_PLAN_ID: &str = "rest.scan.plan-id"; |
There was a problem hiding this comment.
Server-side planning isn't supported yet so this will always be empty anyway. I don't think it's a problem to keep it until it eventually is supported
| let config = response | ||
| .config | ||
| .into_iter() | ||
| .chain(self.user_config.props.clone()) | ||
| .collect(); | ||
| let file_io = self | ||
| .load_file_io(Some(metadata_location), Some(config)) | ||
| .await?; |
There was a problem hiding this comment.
Drive-by fix since this is inconsistent with create_table() and load_table()
| [ | ||
| "auth", | ||
| "token", | ||
| "secret", | ||
| "key", | ||
| "password", | ||
| "cookie", | ||
| "credential", | ||
| ] |
There was a problem hiding this comment.
Not sure if we want to do it like this since I guess there could technically be other headers that would be made sensitive. The main change I wanted to do was also make x-client-secret and x-client-credential sensitive. I think Java redacts every header anyway, so I'm not sure if this really matters
There was a problem hiding this comment.
Given your own note that Java redacts every header regardless of sensitivity, that seems like the simpler and more conservative choice here too. It avoids maintaining a keyword list that can both over-match (a header like x-auth-region gets redacted for no reason) and under-match (a secret header whose name does not contain any of the chosen substrings). Suggest switching to redact-everything-by-default unless there is a concrete case where seeing an unredacted non-sensitive header in error logs matters.
| http = { workspace = true } | ||
| iceberg = { workspace = true } | ||
| itertools = { workspace = true } | ||
| rand = { workspace = true } |
There was a problem hiding this comment.
Not sure if we're okay with adding a new dep rand was already a workspace dep. It's only added for the jittering functionality when retrying
| /// only re-fetch when the current credential is at or near expiry; otherwise | ||
| /// every object-store request would trigger a call back to the catalog. | ||
| #[async_trait] | ||
| pub trait StorageCredentialProvider: Debug + Send + Sync { |
There was a problem hiding this comment.
This is probably one of the most important additions in the PR but it doesn't really follow Java since Java doesn't have an interface over vended credential providers. I think it's probably fine and nicer like this, especially since OpenDAL abstracts everything away quite nicely
There was a problem hiding this comment.
The trait-based design seems fine, Java not having an equivalent is not a reason to avoid one here. See finding 5 above for the separate, concrete question on the same code (pub fields, no constructor).
| if let Some(no_auth) = m.remove(GCS_NO_AUTH) | ||
| && is_truthy(no_auth.to_lowercase().as_str()) | ||
| { |
There was a problem hiding this comment.
Drive-by fix since this looked pretty bad. AWS did this correctly, but GCS would disable this even if gcs.no-auth was set to true
| let url = url::Url::parse(path).map_err(|e| { | ||
| Error::new( | ||
| ErrorKind::DataInvalid, | ||
| format!("Invalid gcs url: {path}: {e}"), | ||
| ) | ||
| })?; |
There was a problem hiding this comment.
S3 had this validation before above but GCS didn't, so another drive-by fix
| if self.uses_dynamic_credentials(&path) { | ||
| let (op, relative_path) = self.create_operator(&path)?; | ||
| op.delete(relative_path).await.map_err(from_opendal_error)?; | ||
| continue; | ||
| } |
There was a problem hiding this comment.
This will probably need to be looked at again later since we forgo batching for paths with a vended credential provider. We would probably have to batch by path-prefix to make this work later. Happy to do this later in this PR or in a new PR
There was a problem hiding this comment.
Agree this should be fixed in this PR rather than deferred. See finding 1 above for the specific fix and the added detail on create_operator also being rebuilt per file.
| /// `reqsign` [`Timestamp`](reqsign_core::time::Timestamp) used on backend | ||
| /// credential types (e.g. `AwsCredential::expires_in`, `google::Token::expires_at`). | ||
| #[cfg(any(feature = "opendal-s3", feature = "opendal-gcs"))] | ||
| pub(crate) fn system_time_to_timestamp( |
There was a problem hiding this comment.
Didn't want to rely on directly on reqsign's Timestamp, but I'm happy to hear different opinions
| /// It contains the location schemes it backs, the property keys it is configured | ||
| /// with, and how to parse its credential. The generic provider stays free of | ||
| /// any per-cloud knowledge. | ||
| struct CloudRefresh { |
There was a problem hiding this comment.
This could also be made into a trait I guess, and we could split aws and gcp support into different modules, but I thought it was small enough to keep it as a struct and make consts for the different cloud providers
| schemes: &["gs", "gcs"], | ||
| endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT, | ||
| enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED, | ||
| jitter_prefetch: false, |
There was a problem hiding this comment.
GCP doesn't have jitter for Java as it's SDK dependent, which only AWS does it appears. I'm happy to discuss more about the retry strategy here in order to make it consistent (which Java doesn't have at the moment)
There was a problem hiding this comment.
Checked this against Java directly. Java's scheduled refresh (S3FileIO/GCSFileIO.refreshStorageCredentials()) never retries after a failed fetch at all, it logs a warning and background refresh permanently stops for that FileIO instance until something else rebuilds the client map. The jittered backoff-and-retry here is already strictly better than Java's behavior, not something that needs to move toward parity. Might be worth saying so explicitly in the module doc comment, since it currently frames the design as "preserving Java's per-cloud refresh policy," and a reader comparing against Java will notice the retry model does not match 1:1.
| } | ||
|
|
||
| /// Resolve a possibly-relative refresh endpoint against the catalog base URI. | ||
| fn resolve_endpoint(base_uri: &str, endpoint: &str) -> String { |
There was a problem hiding this comment.
The behaviour here was done to match Java
|
Hi @mbutrovich! Would it be possible to get a first-round review of this PR when you have time please? |
Yep, it's in my queue! Just slammed with review requests :( Thanks for your patience! |
mbutrovich
left a comment
mbutrovich
left a comment
There was a problem hiding this comment.
First pass, thanks for tackling this @zakariya-s. I have specific feedback, and also some ideas how we might break this up for other reviewers since this is a lot to review in one pass. A split along the layers already present in the diff looks fairly clean and, other than one ordering constraint, each piece is independently testable rather than a bare stub:
- The
Debug-redaction hardening (RestCatalogConfig,HttpClient,StorageConfig,LoadTableResult,StorageCredential,OpenDalResolvingStorage, plus theis_sensitive_headerbroadening) and the two small unrelated fixes riding along (theregister_tableconfig merge, the GCSno-authtruthy parsing). None of this depends on credential refresh existing, and it already has its own tests. - The
StorageCredentialProvidertrait and credential types in theicebergcrate, plusStorageFactory::build_with_credentialswith its safe default (errors if a provider is supplied and the factory has not opted in). Covered by finding 5. No behavior change for existing backends. - OpenDAL S3 consuming the trait (the adapter, the anonymous-access guard, the
delete_streamchange). Testable on its own with a hand-rolled provider, same as the tests already in this diff, without needing anything from the REST side. - OpenDAL GCS consuming the trait, same shape as 3, independent of it.
Finding 1 (lost delete batching) sits in the shared uses_dynamic_credentials/delete_stream code in lib.rs rather than in s3.rs or gcs.rs specifically, so it isn't purely a 3-or-4 problem: whichever of the two lands first introduces that shared mechanism, and the other reuses it as-is, so the fix only needs to happen once.
- The REST catalog fetch/cache/jitter/backoff logic (
RestVendedCredentialProvider,CloudRefresh,resolve_endpoint, the table-scoped clientfor_table). Covered by findings 2, 3, 4, and the root cause of 6 (the fresh-HttpClient-per-table-scoped-provider behavior lives inclient.rs/credential.rs, both part of this PR). This can be written and tested against the trait directly withmockito, same as the tests already in this diff, without touching OpenDAL at all. - Wiring
RestCatalog::load_file_ioto actually attach the provider (catalog.rs:561-565). Covered by finding 7 (the inline path at that call site) and the rest of finding 6 (this is the call site that turns "everyload_file_iocall re-runs the OAuth handshake" from a latent property offor_tableinto something that fires on everyload_table/create_table/register_table).
This piece should land last on purpose, not just for tidiness: client.refresh-credentials-endpoint and gcs.oauth2.refresh-credentials-enabled are property keys the Java client and the REST spec already define, so a production catalog that's Java-interoperable may already be sending them today regardless of which client is asking. If the wiring lands before both the S3 and GCS sides can consume a provider, every existing rust client hitting such a catalog would start failing table loads on S3 or GCS the moment that PR merges, since the default build_with_credentials errors whenever a provider is supplied to a factory that hasn't opted in. So 3 and 4 both need to land before 6, even though 3, 4, and 5 are otherwise independent of each other and can be reviewed in any order.
#2931 is currently a single feature request rather than a tracking issue for a multi-PR stack. Worth turning it into one, or opening a separate tracking issue with a checklist for the pieces above, so reviewers can see the whole plan and where a given PR sits in it before reviewing any single piece.
| fn uses_dynamic_credentials(&self, path: &str) -> bool { | ||
| match self { | ||
| #[cfg(feature = "opendal-s3")] | ||
| OpenDalStorage::S3 { | ||
| credential_provider: Some(provider), | ||
| .. | ||
| } => provider.supports_path(path), | ||
| #[cfg(feature = "opendal-gcs")] | ||
| OpenDalStorage::Gcs { |
There was a problem hiding this comment.
crates/storage/opendal/src/lib.rs:423-431 (uses_dynamic_credentials) and :610-628 (delete_stream).
When a path is served by a credential provider, delete_stream skips the shared per-bucket Deleter and instead calls create_operator + a single op.delete(relative_path) per path, sequentially, inside the stream loop. The non-dynamic branch batches deletes through OpenDAL's Deleter (which can use bulk delete APIs); the dynamic branch does neither batching nor concurrency, and rebuilds the operator from scratch for every single file.
For expire_snapshots/purge on a table with vended-credential refresh enabled, this turns what would be a handful of batched multi-object delete calls into one HTTP round trip per file, plus an operator-construction cost per file. The code comment explains why deletes can't share a Deleter across different credential-prefix scopes (correctness: batch_key_for_path only groups by bucket, not by credential scope), but the fix taken forfeits batching entirely rather than partially, i.e. grouping deletes by (bucket, matched credential prefix) instead of just bucket would preserve batched delete within each credential-scope group. Was that considered?
| let refreshed = self.fetch(configured).await.and_then(|entries| { | ||
| let credential = longest_prefix_match(&entries, path) | ||
| .filter(|entry| entry.is_unexpired(SystemTime::now())) | ||
| .map(|entry| entry.credential.clone()) | ||
| .ok_or_else(|| { | ||
| Error::new( | ||
| ErrorKind::Unexpected, | ||
| format!("no unexpired vended credential matches storage location: {path}"), | ||
| ) | ||
| })?; | ||
| Ok((entries, credential)) | ||
| }); | ||
|
|
||
| match refreshed { | ||
| Ok((entries, credential)) => { | ||
| let mut cache = configured.cache.lock().await; | ||
| cache.entries = entries; | ||
| cache.consecutive_failures = 0; | ||
| cache.retry_not_before = None; | ||
| Ok(credential) | ||
| } | ||
| Err(fetch_error) => { | ||
| let mut cache = configured.cache.lock().await; | ||
| cache.consecutive_failures = cache.consecutive_failures.saturating_add(1); | ||
| cache.retry_not_before = | ||
| Instant::now().checked_add(failure_backoff(cache.consecutive_failures)); | ||
|
|
||
| // Graceful degradation: while the cached credential remains | ||
| // usable, serve it and retry after jittered backoff. Expired | ||
| // credentials are never served. | ||
| fallback | ||
| .filter(|entry| entry.is_unexpired(SystemTime::now())) | ||
| .map(|entry| entry.credential) | ||
| .ok_or(fetch_error) | ||
| } | ||
| } |
There was a problem hiding this comment.
crates/catalog/rest/src/credential.rs:265-300 (refresh_credential), contrast with S3FileIO.refreshStorageCredentials()/GCSFileIO.refreshStorageCredentials() in Java.
Java's actual multi-prefix refresh (used by both S3FileIO and GCSFileIO, not the single-credential VendedCredentialsProvider/OAuth2RefreshCredentialsHandler used as an SDK credentials provider) is unconditional: on each scheduled refresh it fetches the credentials endpoint once, keeps every entry matching the cloud's root prefix, and replaces storageCredentials wholesale — no per-path filtering happens at refresh time at all.
The Rust refresh_credential instead does per-path filtering inline: it fetches all entries for a cloud, then immediately narrows to longest_prefix_match(&entries, path).filter(unexpired) for this specific call's path. If that narrowing yields nothing (no entry covers this path, or the covering entry happens to already be expired), the whole outcome is treated as Err and:
- the freshly-fetched
entriesare never written tocache.entries(only theOk((entries, credential))branch at line 279-284 updates the cache) — so if the response contained entries for other prefixes (as the existinglongest_prefix_match_ignores_freshnesstest exercises), they're thrown away even though a subsequent call for a different, valid path would have to fetch them all over again; cache.consecutive_failuresis incremented andretry_not_beforebackoff is armed (lines 288-290) even though the catalog responded successfully — it just didn't vend anything for this path.
Given Java's model of "cache everything the endpoint returns, unconditionally," was per-path filtering at refresh time (rather than only at cache-read time, where it already happens in cache_decision/longest_prefix_match) intentional here?
| parsed | ||
| .storage_credentials | ||
| .into_iter() | ||
| .filter(|sc| configured.cloud.matches_location(&sc.prefix)) | ||
| .map(|sc| { | ||
| (configured.cloud.parse_credential)(&sc.config, Some(sc.prefix)).map( | ||
| |credential| { | ||
| CachedEntry::new(credential, configured.cloud.jitter_prefetch) | ||
| }, | ||
| ) | ||
| }) | ||
| .collect() |
There was a problem hiding this comment.
collect() into Result<Vec<CachedEntry>> short-circuits on the first parse_credential error. If a server vends N credentials for one cloud and one entry is missing a required field, all N become unusable (and, per finding 2, this also counts as a "failure" for backoff purposes) rather than just the one bad entry. Java's equivalent (VendedCredentialsProvider.refreshCredential) only ever expects a single S3-prefixed entry and asserts on it directly, so there's no directly analogous "partial batch" behavior to compare against — but given this PR's own design supports N entries per cloud, is one bad entry meant to invalidate all the others?
| let enabled = props | ||
| .get(cloud.enabled_key) | ||
| .is_none_or(|value| value.parse().unwrap_or(false)); |
There was a problem hiding this comment.
str::parse::<bool>() only accepts the exact strings "true"/"false". Java's PropertyUtil.propertyAsBoolean uses Boolean.parseBoolean, which is case-insensitive for "true" ("True", "TRUE" all parse as true). A config value of client.refresh-credentials-enabled: "True" would enable refresh in Java but silently disable it in Rust (parse error → unwrap_or(false)). Low severity (fails closed either way), but worth a case-insensitive comparison to match Java's actual accepted input space.
| pub struct StorageCredential { | ||
| /// Storage-location prefix this credential is scoped to. `None` represents a | ||
| /// credential without a declared scope, sourced from flat storage properties. | ||
| pub prefix: Option<String>, | ||
| /// The backend-specific credential material. | ||
| pub kind: StorageCredentialKind, | ||
| /// When the credential expires, if known. `None` means non-expiring and | ||
| /// backends treat such a credential as always valid and never refresh it. | ||
| pub expires_at: Option<SystemTime>, | ||
| } | ||
|
|
||
| /// Backend-specific credential material. | ||
| #[derive(Clone, Debug)] | ||
| pub enum StorageCredentialKind { | ||
| /// Amazon S3 credentials. | ||
| S3(S3Credential), | ||
| /// Google Cloud Storage credentials. | ||
| Gcs(GcsCredential), | ||
| } | ||
|
|
||
| /// Temporary Amazon S3 credentials. | ||
| #[derive(Clone)] | ||
| pub struct S3Credential { | ||
| /// AWS access key ID. | ||
| pub access_key_id: String, | ||
| /// AWS secret access key. | ||
| pub secret_access_key: String, | ||
| /// AWS session token, set for temporary (STS/vended) credentials. | ||
| pub session_token: Option<String>, | ||
| } | ||
|
|
||
| impl Debug for S3Credential { | ||
| fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { | ||
| f.debug_struct("S3Credential").finish_non_exhaustive() | ||
| } | ||
| } | ||
|
|
||
| /// Temporary Google Cloud Storage credentials (an OAuth2 access token). | ||
| #[derive(Clone)] | ||
| pub struct GcsCredential { | ||
| /// OAuth2 bearer token used to access GCS. | ||
| pub token: String, | ||
| } |
There was a problem hiding this comment.
These are new public types (in public-api.txt) that third-party StorageCredentialProvider implementors must construct by hand. Every field is pub, with no constructor. That's inconsistent with StorageConfig in the same module (crates/iceberg/src/io/storage/config/mod.rs:55-58), which keeps props private and exposes with_prop/from_props instead. Was a constructor considered, or is direct struct-literal construction intentional here?
| let file_io = FileIOBuilder::new(factory).with_props(props).build(); | ||
| // If the catalog vends refreshable credentials for this table's storage, | ||
| // attach a provider so the backend re-fetches them before they expire. | ||
| let credential_provider = crate::credential::build_vended_credential_provider( |
There was a problem hiding this comment.
The file already has a use crate::client::{...} / use crate::types::{...} block at the top; this call could join it as use crate::credential::build_vended_credential_provider; instead of qualifying the path inline at the call site.
| if self.uses_dynamic_credentials(&path) { | ||
| let (op, relative_path) = self.create_operator(&path)?; | ||
| op.delete(relative_path).await.map_err(from_opendal_error)?; | ||
| continue; | ||
| } |
There was a problem hiding this comment.
Agree this should be fixed in this PR rather than deferred. See finding 1 above for the specific fix and the added detail on create_operator also being rebuilt per file.
| schemes: &["gs", "gcs"], | ||
| endpoint_key: GCS_REFRESH_CREDENTIALS_ENDPOINT, | ||
| enabled_key: GCS_REFRESH_CREDENTIALS_ENABLED, | ||
| jitter_prefetch: false, |
There was a problem hiding this comment.
Checked this against Java directly. Java's scheduled refresh (S3FileIO/GCSFileIO.refreshStorageCredentials()) never retries after a failed fetch at all, it logs a warning and background refresh permanently stops for that FileIO instance until something else rebuilds the client map. The jittered backoff-and-retry here is already strictly better than Java's behavior, not something that needs to move toward parity. Might be worth saying so explicitly in the module doc comment, since it currently frames the design as "preserving Java's per-cloud refresh policy," and a reader comparing against Java will notice the retry model does not match 1:1.
| [ | ||
| "auth", | ||
| "token", | ||
| "secret", | ||
| "key", | ||
| "password", | ||
| "cookie", | ||
| "credential", | ||
| ] |
There was a problem hiding this comment.
Given your own note that Java redacts every header regardless of sensitivity, that seems like the simpler and more conservative choice here too. It avoids maintaining a keyword list that can both over-match (a header like x-auth-region gets redacted for no reason) and under-match (a secret header whose name does not contain any of the chosen substrings). Suggest switching to redact-everything-by-default unless there is a concrete case where seeing an unredacted non-sensitive header in error logs matters.
| /// only re-fetch when the current credential is at or near expiry; otherwise | ||
| /// every object-store request would trigger a call back to the catalog. | ||
| #[async_trait] | ||
| pub trait StorageCredentialProvider: Debug + Send + Sync { |
There was a problem hiding this comment.
The trait-based design seems fine, Java not having an equivalent is not a reason to avoid one here. See finding 5 above for the separate, concrete question on the same code (pub fields, no constructor).
2 participants
Which issue does this PR close?
What changes are included in this PR?
This PR adds support for refreshing short-lived storage credentials vended by REST catalogs:
StorageCredentialProviderinterface toFileIOheader.*properties for refresh requestsDebugoutputAzure credential refresh is not included because the current OpenDAL Azure backend does not expose the credential-provider and expiry hooks required for safe refresh.
Are these changes tested?
Yes