plusplusjiajia commented on code in PR #892:
URL: https://github.com/apache/iceberg-cpp/pull/892#discussion_r4256411700
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -235,14 +306,60 @@ class ArrowS3FileIO final : public FileIO, public
SupportsStorageCredentials {
void InstallCredentials(std::vector<StorageCredential>& storage_credentials,
DelegatesByPrefix& delegates);
+ /// \brief Whether the installed credentials are close enough to expiring to
+ /// be replaced, and no backoff is in effect.
+ ///
+ /// Callers must hold `mutex_`, at least shared.
+ bool RefreshDue() const;
+
+ /// \brief Whether the installed credentials have already stopped being
valid.
+ ///
+ /// Callers must hold `mutex_`, at least shared.
+ bool Expired() const;
+
+ /// \brief When the next refresh attempt becomes allowed after a failure.
+ ///
+ /// Never past the point the credentials stop being valid.
+ ///
+ /// Callers must hold `mutex_`, at least shared.
+ std::chrono::steady_clock::time_point BackoffUntil() const;
+
+ /// \brief Replace the credentials once they are close to expiring.
+ ///
+ /// Called before each handle is created; a handle keeps the delegate it was
+ /// built from, so I/O on an open one is not re-checked. A failure keeps the
Review Comment:
Thanks! Done. Each vended S3 client now signs through a
`VendedCredentialsProvider` set as Arrow's `S3Options::credentials_provider`.
The AWS SDK asks it before every request: it returns the keys the FileIO
currently vends for the client's prefix and refreshes them once due, so
existing handles and streams pick up new tokens. A token-only refresh keeps the
existing clients; only a change of prefixes or settings builds new ones. One
caveat: Arrow holds its client registry while a request runs, and building a
client locks it exclusively, so a refresh started from a request that needs new
clients is left to the next file operation.
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -291,9 +407,131 @@ Result<ArrowS3FileIO::DelegatesByPrefix>
ArrowS3FileIO::BuildDelegates(
void ArrowS3FileIO::InstallCredentials(
std::vector<StorageCredential>& storage_credentials, DelegatesByPrefix&
delegates) {
file_io_by_prefix_.swap(delegates);
+ expires_at_ = EarliestExpiry(storage_credentials);
+ retry_refresh_at_ = {};
+ ++credential_generation_;
storage_credentials_.swap(storage_credentials);
}
+bool ArrowS3FileIO::RefreshDue() const {
+ return expires_at_.has_value() &&
+ std::chrono::system_clock::now() + kRefreshLeadTime >= *expires_at_ &&
+ std::chrono::steady_clock::now() >= retry_refresh_at_;
+}
+
+bool ArrowS3FileIO::Expired() const {
+ return expires_at_.has_value() && std::chrono::system_clock::now() >=
*expires_at_;
+}
+
+std::chrono::steady_clock::time_point ArrowS3FileIO::BackoffUntil() const {
+ auto delay =
+
std::chrono::duration_cast<std::chrono::milliseconds>(kRefreshRetryBackoff);
+ if (expires_at_.has_value()) {
+ const auto remaining =
std::chrono::duration_cast<std::chrono::milliseconds>(
+ *expires_at_ - std::chrono::system_clock::now());
+ // Worth retrying before they run out; once they have, faster retries only
+ // hammer a catalog that is already failing.
+ if (remaining > std::chrono::milliseconds::zero()) {
+ delay = std::clamp(
+ remaining,
+
std::chrono::duration_cast<std::chrono::milliseconds>(kMinRefreshRetryBackoff),
+ delay);
+ }
+ }
+ return std::chrono::steady_clock::now() + delay;
+}
+
+void ArrowS3FileIO::MaybeRefreshCredentials() {
+ {
+ // Cheap pre-check, so the common case costs one shared lock and no more.
+ std::shared_lock lock(mutex_);
+ if (!provider_ || !RefreshDue()) {
+ return;
+ }
+ }
+
+ std::unique_lock refresh_lock(refresh_mutex_, std::defer_lock);
+ if (!refresh_lock.try_lock()) {
+ // Another operation is already fetching; normally just use what we have.
+ {
+ std::shared_lock lock(mutex_);
+ if (!Expired()) {
+ return;
+ }
+ }
+ // Expired credentials leave nothing to proceed with, so wait instead.
+ if (!refresh_lock.try_lock_for(kExpiredCredentialWait)) {
+ return;
Review Comment:
Thanks! Done. Each prefix's delegate now tracks its own expiry. Once it has
passed and the refresh failed or the wait timed out, creating a handle or
deleting under that prefix returns `TokenExpired`. Locations outside vended
prefixes keep the default credentials.
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -291,9 +407,131 @@ Result<ArrowS3FileIO::DelegatesByPrefix>
ArrowS3FileIO::BuildDelegates(
void ArrowS3FileIO::InstallCredentials(
std::vector<StorageCredential>& storage_credentials, DelegatesByPrefix&
delegates) {
file_io_by_prefix_.swap(delegates);
+ expires_at_ = EarliestExpiry(storage_credentials);
+ retry_refresh_at_ = {};
+ ++credential_generation_;
storage_credentials_.swap(storage_credentials);
}
+bool ArrowS3FileIO::RefreshDue() const {
+ return expires_at_.has_value() &&
+ std::chrono::system_clock::now() + kRefreshLeadTime >= *expires_at_ &&
+ std::chrono::steady_clock::now() >= retry_refresh_at_;
+}
+
+bool ArrowS3FileIO::Expired() const {
+ return expires_at_.has_value() && std::chrono::system_clock::now() >=
*expires_at_;
+}
+
+std::chrono::steady_clock::time_point ArrowS3FileIO::BackoffUntil() const {
+ auto delay =
+
std::chrono::duration_cast<std::chrono::milliseconds>(kRefreshRetryBackoff);
+ if (expires_at_.has_value()) {
+ const auto remaining =
std::chrono::duration_cast<std::chrono::milliseconds>(
+ *expires_at_ - std::chrono::system_clock::now());
+ // Worth retrying before they run out; once they have, faster retries only
+ // hammer a catalog that is already failing.
+ if (remaining > std::chrono::milliseconds::zero()) {
+ delay = std::clamp(
+ remaining,
+
std::chrono::duration_cast<std::chrono::milliseconds>(kMinRefreshRetryBackoff),
+ delay);
+ }
+ }
+ return std::chrono::steady_clock::now() + delay;
+}
+
+void ArrowS3FileIO::MaybeRefreshCredentials() {
+ {
+ // Cheap pre-check, so the common case costs one shared lock and no more.
+ std::shared_lock lock(mutex_);
+ if (!provider_ || !RefreshDue()) {
+ return;
+ }
+ }
+
+ std::unique_lock refresh_lock(refresh_mutex_, std::defer_lock);
+ if (!refresh_lock.try_lock()) {
+ // Another operation is already fetching; normally just use what we have.
+ {
+ std::shared_lock lock(mutex_);
+ if (!Expired()) {
+ return;
+ }
+ }
+ // Expired credentials leave nothing to proceed with, so wait instead.
+ if (!refresh_lock.try_lock_for(kExpiredCredentialWait)) {
+ return;
+ }
+ }
+ // Snapshot the provider and credentials' generation together.
+ std::shared_ptr<StorageCredentialProvider> provider;
+ uint64_t generation = 0;
+ {
+ // Whoever held the lock may also have just finished, leaving nothing to
do.
+ std::shared_lock lock(mutex_);
+ if (!provider_ || !RefreshDue()) {
+ return;
+ }
+ provider = provider_;
+ generation = credential_generation_;
+ }
+
+ // Outside `mutex_`: both are slow and must not block readers.
+ Status status;
+ DelegatesByPrefix delegates;
+ auto refreshed = provider->Load();
+ if (refreshed.has_value()) {
+ auto built = BuildDelegates(*refreshed);
+ if (!built.has_value()) {
+ status = std::unexpected(built.error());
+ } else if (built->empty()) {
+ // Installing this would drop working credentials for whatever ambient
+ // identity the AWS chain finds. Java refuses an empty list too.
+ status = NotFound("Refreshed credentials contain no S3-compatible
prefix");
+ } else {
+ delegates = std::move(built).value();
+ }
+ } else {
+ status = std::unexpected(refreshed.error());
+ }
+
+ std::unique_lock lock(mutex_);
+ // Credentials installed meanwhile supersede this refresh: what it fetched is
+ // by now the older set.
+ const bool superseded = credential_generation_ != generation;
+ if (!status.has_value()) {
+ // Reported either way, so a failing catalog stays visible.
+ if (superseded) {
+ ICEBERG_LOG_WARN(
+ "Failed to refresh vended storage credentials ({}); they have since
been "
+ "replaced",
+ status.error().message);
+ return;
+ }
+ retry_refresh_at_ = BackoffUntil();
+ ICEBERG_LOG_WARN(
+ "Failed to refresh vended storage credentials ({}); keeping the
current "
+ "ones and retrying in {}ms",
+ status.error().message,
+ std::chrono::duration_cast<std::chrono::milliseconds>(
+ retry_refresh_at_ - std::chrono::steady_clock::now())
+ .count());
+ return;
+ }
+ if (superseded) {
+ return;
+ }
+
+ // `delegates`/`*refreshed` take the retired generation; both are declared
+ // before the lock, so it destructs only after the lock releases.
+ InstallCredentials(*refreshed, delegates);
Review Comment:
Thanks! Done. A refreshed set is installed only if every S3 credential is
usable: a session token needs a parseable expiry that has not passed, and the
set must contain an S3 prefix. Otherwise the current set is kept and the
refresh backs off as for any failed refresh, never past the current expiry. A
token without a usable expiry at the initial install still triggers an
immediate refresh, but is no longer treated as expired.
##########
src/iceberg/arrow/s3/arrow_s3_file_io.cc:
##########
@@ -291,9 +407,131 @@ Result<ArrowS3FileIO::DelegatesByPrefix>
ArrowS3FileIO::BuildDelegates(
void ArrowS3FileIO::InstallCredentials(
std::vector<StorageCredential>& storage_credentials, DelegatesByPrefix&
delegates) {
file_io_by_prefix_.swap(delegates);
+ expires_at_ = EarliestExpiry(storage_credentials);
+ retry_refresh_at_ = {};
+ ++credential_generation_;
storage_credentials_.swap(storage_credentials);
}
+bool ArrowS3FileIO::RefreshDue() const {
+ return expires_at_.has_value() &&
+ std::chrono::system_clock::now() + kRefreshLeadTime >= *expires_at_ &&
+ std::chrono::steady_clock::now() >= retry_refresh_at_;
+}
+
+bool ArrowS3FileIO::Expired() const {
+ return expires_at_.has_value() && std::chrono::system_clock::now() >=
*expires_at_;
+}
+
+std::chrono::steady_clock::time_point ArrowS3FileIO::BackoffUntil() const {
+ auto delay =
+
std::chrono::duration_cast<std::chrono::milliseconds>(kRefreshRetryBackoff);
+ if (expires_at_.has_value()) {
+ const auto remaining =
std::chrono::duration_cast<std::chrono::milliseconds>(
+ *expires_at_ - std::chrono::system_clock::now());
+ // Worth retrying before they run out; once they have, faster retries only
+ // hammer a catalog that is already failing.
+ if (remaining > std::chrono::milliseconds::zero()) {
+ delay = std::clamp(
+ remaining,
+
std::chrono::duration_cast<std::chrono::milliseconds>(kMinRefreshRetryBackoff),
+ delay);
+ }
+ }
+ return std::chrono::steady_clock::now() + delay;
+}
+
+void ArrowS3FileIO::MaybeRefreshCredentials() {
+ {
+ // Cheap pre-check, so the common case costs one shared lock and no more.
+ std::shared_lock lock(mutex_);
+ if (!provider_ || !RefreshDue()) {
+ return;
+ }
+ }
+
+ std::unique_lock refresh_lock(refresh_mutex_, std::defer_lock);
+ if (!refresh_lock.try_lock()) {
+ // Another operation is already fetching; normally just use what we have.
+ {
+ std::shared_lock lock(mutex_);
+ if (!Expired()) {
+ return;
+ }
+ }
+ // Expired credentials leave nothing to proceed with, so wait instead.
+ if (!refresh_lock.try_lock_for(kExpiredCredentialWait)) {
+ return;
+ }
+ }
+ // Snapshot the provider and credentials' generation together.
+ std::shared_ptr<StorageCredentialProvider> provider;
+ uint64_t generation = 0;
+ {
+ // Whoever held the lock may also have just finished, leaving nothing to
do.
+ std::shared_lock lock(mutex_);
+ if (!provider_ || !RefreshDue()) {
+ return;
+ }
+ provider = provider_;
+ generation = credential_generation_;
+ }
+
+ // Outside `mutex_`: both are slow and must not block readers.
+ Status status;
+ DelegatesByPrefix delegates;
+ auto refreshed = provider->Load();
Review Comment:
Thanks! Done. `HttpClient::Get` takes an optional timeout, and
`RestStorageCredentialProvider::Load()` uses 10 seconds, so a stalled catalog
no longer blocks the refreshing operation indefinitely; a timeout counts as a
failed 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
