Skip to content

Commit 7ccee12

Browse files
feat(rest): load table credentials from the LoadCredentials endpoint
1 parent 896b3ce commit 7ccee12

14 files changed

Lines changed: 305 additions & 27 deletions

‎src/iceberg/catalog/rest/json_serde.cc‎

Lines changed: 33 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,25 @@ Result<StorageCredential> StorageCredentialFromJson(const nlohmann::json& json)
152152
return credential;
153153
}
154154

155+
/// \brief Reads the optional `storage-credentials` array shared by the
156+
/// LoadTable and LoadCredentials responses.
157+
Result<std::vector<StorageCredential>> StorageCredentialsFromJson(
158+
const nlohmann::json& json) {
159+
std::vector<StorageCredential> credentials;
160+
auto it = json.find(kStorageCredentials);
161+
if (it == json.end() || it->is_null()) {
162+
return credentials;
163+
}
164+
if (!it->is_array()) {
165+
return JsonParseError("Cannot parse storage credentials from non-array");
166+
}
167+
for (const auto& entry : *it) {
168+
ICEBERG_ASSIGN_OR_RAISE(auto credential, StorageCredentialFromJson(entry));
169+
credentials.push_back(std::move(credential));
170+
}
171+
return credentials;
172+
}
173+
155174
template <typename Value>
156175
Result<std::map<int32_t, Value>> KeyValueMapFromJson(const nlohmann::json& json,
157176
std::string_view key) {
@@ -738,19 +757,24 @@ Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
738757
ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
739758
ICEBERG_ASSIGN_OR_RAISE(result.config,
740759
GetJsonValueOrDefault<decltype(result.config)>(json, kConfig));
741-
if (auto it = json.find(kStorageCredentials); it != json.end() && !it->is_null()) {
742-
if (!it->is_array()) {
743-
return JsonParseError("Cannot parse storage credentials from non-array");
744-
}
745-
for (const auto& entry : *it) {
746-
ICEBERG_ASSIGN_OR_RAISE(auto cred, StorageCredentialFromJson(entry));
747-
result.storage_credentials.push_back(std::move(cred));
748-
}
749-
}
760+
ICEBERG_ASSIGN_OR_RAISE(result.storage_credentials, StorageCredentialsFromJson(json));
750761
ICEBERG_RETURN_UNEXPECTED(result.Validate());
751762
return result;
752763
}
753764

765+
Result<LoadCredentialsResponse> LoadCredentialsResponseFromJson(
766+
const nlohmann::json& json) {
767+
// Required here, unlike in LoadTable: reading a malformed response as "no
768+
// credentials" would look like a refresh that succeeded and dropped them.
769+
if (auto it = json.find(kStorageCredentials); it == json.end() || it->is_null()) {
770+
return JsonParseError("Missing '{}'", kStorageCredentials);
771+
}
772+
LoadCredentialsResponse response;
773+
ICEBERG_ASSIGN_OR_RAISE(response.storage_credentials, StorageCredentialsFromJson(json));
774+
ICEBERG_RETURN_UNEXPECTED(response.Validate());
775+
return response;
776+
}
777+
754778
nlohmann::json ToJson(const ListNamespacesResponse& response) {
755779
nlohmann::json json;
756780
SetOptionalStringField(json, kNextPageToken, response.next_page_token);

‎src/iceberg/catalog/rest/json_serde_internal.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,10 @@ template <>
7171
ICEBERG_REST_EXPORT Result<LoadTableResult> FromJson(const nlohmann::json& json);
7272
ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(const LoadTableResult& model);
7373

74+
// Response-only model: a client never serializes it, so no ToJson.
75+
ICEBERG_REST_EXPORT Result<LoadCredentialsResponse> LoadCredentialsResponseFromJson(
76+
const nlohmann::json& json);
77+
7478
ICEBERG_REST_EXPORT Result<CreateTableRequest> CreateTableRequestFromJson(
7579
const nlohmann::json& json);
7680
template <>

‎src/iceberg/catalog/rest/rest_catalog.cc‎

Lines changed: 63 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
#include "iceberg/catalog/rest/rest_util.h"
4343
#include "iceberg/catalog/rest/types.h"
4444
#include "iceberg/json_serde_internal.h"
45+
#include "iceberg/logging/log_macros.h"
4546
#include "iceberg/metrics/metrics_reporters.h"
4647
#include "iceberg/partition_spec.h"
4748
#include "iceberg/result.h"
@@ -508,12 +509,65 @@ Result<std::shared_ptr<auth::AuthSession>> RestCatalog::TableAuthSession(
508509
std::move(contextual_session));
509510
}
510511

512+
StorageCredentialRefresher RestCatalog::MakeCredentialRefresher(
513+
const TableIdentifier& identifier,
514+
std::shared_ptr<auth::AuthSession> table_session) const {
515+
if (!supported_endpoints_.contains(Endpoint::TableCredentials())) {
516+
// Not an error, but it surfaces much later as credentials expiring.
517+
ICEBERG_LOG_DEBUG(
518+
"Catalog does not advertise {}; vended credentials for '{}' will not be "
519+
"refreshed",
520+
Endpoint::TableCredentials().ToString(), ToString(identifier));
521+
return nullptr;
522+
}
523+
auto path = paths_->Credentials(identifier);
524+
if (!path.has_value()) {
525+
ICEBERG_LOG_WARN(
526+
"Cannot build the credentials path for '{}' ({}); its vended credentials "
527+
"will not be refreshed",
528+
ToString(identifier), path.error().message);
529+
return nullptr;
530+
}
531+
auto client = client_;
532+
auto credentials_path = std::move(path.value());
533+
auto session = std::move(table_session);
534+
// The catalog's destructor closes the session, and a table's FileIO can
535+
// outlive the table keeping the catalog alive. No cycle: the catalog's own
536+
// FileIO never gets a refresher.
537+
auto catalog = shared_from_this();
538+
return [catalog, client, credentials_path,
539+
session]() -> Result<std::vector<StorageCredential>> {
540+
ICEBERG_ASSIGN_OR_RAISE(const auto response,
541+
client->Get(credentials_path, /*params=*/{}, /*headers=*/{},
542+
*TableErrorHandler::Instance(), *session));
543+
// Parse errors embed the offending input, and this body carries
544+
// credentials; strip the message so it can never reach a log.
545+
auto json = FromJsonString(response.body());
546+
if (!json.has_value()) {
547+
return JsonParseError("Malformed LoadCredentials response");
548+
}
549+
auto result = LoadCredentialsResponseFromJson(*json);
550+
if (!result.has_value()) {
551+
return std::unexpected<Error>(
552+
{.kind = result.error().kind, .message = "Malformed LoadCredentials response"});
553+
}
554+
return std::move(result->storage_credentials);
555+
};
556+
}
557+
511558
Result<std::shared_ptr<FileIO>> RestCatalog::TableFileIO(
512-
const SessionContext& /*context*/,
559+
const SessionContext& /*context*/, const TableIdentifier& identifier,
513560
const std::unordered_map<std::string, std::string>& table_config,
514-
const std::vector<StorageCredential>& storage_credentials) const {
561+
const std::vector<StorageCredential>& storage_credentials,
562+
std::shared_ptr<auth::AuthSession> table_session) const {
515563
if (!table_config.empty() || !storage_credentials.empty()) {
516-
return MakeTableFileIO(config_.configs(), table_config, storage_credentials);
564+
// Only vended credentials expire, so only they need a refresher.
565+
StorageCredentialRefresher refresher;
566+
if (!storage_credentials.empty()) {
567+
refresher = MakeCredentialRefresher(identifier, std::move(table_session));
568+
}
569+
return MakeTableFileIO(config_.configs(), table_config, storage_credentials,
570+
std::move(refresher));
517571
}
518572

519573
return file_io_;
@@ -772,11 +826,12 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
772826
/*stage_create=*/true, *contextual_session));
773827
auto table_config = std::move(result.config);
774828
auto storage_credentials = std::move(result.storage_credentials);
775-
ICEBERG_ASSIGN_OR_RAISE(auto table_io,
776-
TableFileIO(context, table_config, storage_credentials));
829+
// Before the FileIO: refreshing its credentials reuses the table session.
777830
ICEBERG_ASSIGN_OR_RAISE(
778831
auto table_session,
779832
TableAuthSession(identifier, table_config, std::move(contextual_session)));
833+
ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, identifier, table_config,
834+
storage_credentials, table_session));
780835
ICEBERG_ASSIGN_OR_RAISE(auto reporter, MakeTableReporter(identifier, table_session));
781836
auto table_catalog = std::make_shared<TableScopedCatalog>(
782837
shared_from_this(), context, identifier, table_config, std::move(table_session),
@@ -890,11 +945,12 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
890945
std::shared_ptr<auth::AuthSession> contextual_session) {
891946
auto table_config = std::move(result.config);
892947
auto storage_credentials = std::move(result.storage_credentials);
893-
ICEBERG_ASSIGN_OR_RAISE(auto table_io,
894-
TableFileIO(context, table_config, storage_credentials));
948+
// Before the FileIO: refreshing its credentials reuses the table session.
895949
ICEBERG_ASSIGN_OR_RAISE(
896950
auto table_session,
897951
TableAuthSession(identifier, table_config, std::move(contextual_session)));
952+
ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, identifier, table_config,
953+
storage_credentials, table_session));
898954
ICEBERG_ASSIGN_OR_RAISE(auto reporter, MakeTableReporter(identifier, table_session));
899955
auto table_catalog = std::make_shared<TableScopedCatalog>(
900956
shared_from_this(), context, identifier, table_config, table_session, table_io);

‎src/iceberg/catalog/rest/rest_catalog.h‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -85,9 +85,19 @@ class ICEBERG_REST_EXPORT RestCatalog final
8585
std::shared_ptr<auth::AuthSession> contextual_session);
8686

8787
Result<std::shared_ptr<FileIO>> TableFileIO(
88-
const SessionContext& context,
88+
const SessionContext& context, const TableIdentifier& identifier,
8989
const std::unordered_map<std::string, std::string>& table_config,
90-
const std::vector<StorageCredential>& storage_credentials) const;
90+
const std::vector<StorageCredential>& storage_credentials,
91+
std::shared_ptr<auth::AuthSession> table_session) const;
92+
93+
/// \brief Callback that reloads this table's vended credentials, or nullptr
94+
/// when the catalog does not serve the LoadCredentials endpoint.
95+
///
96+
/// Captures the catalog, so anything documented to outlive it -- the metrics
97+
/// executor included -- must also outlive every FileIO holding a refresher.
98+
StorageCredentialRefresher MakeCredentialRefresher(
99+
const TableIdentifier& identifier,
100+
std::shared_ptr<auth::AuthSession> table_session) const;
91101

92102
Result<std::vector<Namespace>> ListNamespaces(const Namespace& ns,
93103
auth::AuthSession& session) const;

‎src/iceberg/catalog/rest/rest_file_io.cc‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121

2222
#include <string>
2323
#include <unordered_map>
24+
#include <utility>
2425
#include <vector>
2526

2627
#include "iceberg/catalog/rest/types.h"
@@ -57,14 +58,19 @@ Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& c
5758
Result<std::unique_ptr<FileIO>> MakeTableFileIO(
5859
const std::unordered_map<std::string, std::string>& catalog_config,
5960
const std::unordered_map<std::string, std::string>& table_config,
60-
const std::vector<StorageCredential>& storage_credentials) {
61+
const std::vector<StorageCredential>& storage_credentials,
62+
StorageCredentialRefresher refresher) {
6163
const auto default_properties = MergeFileIOProperties(catalog_config, table_config);
6264
ICEBERG_ASSIGN_OR_RAISE(
6365
auto io, MakeCatalogFileIO(RestCatalogProperties::FromMap(default_properties)));
6466

6567
if (storage_credentials.empty()) {
6668
return io;
6769
} else if (auto* credentialed = io->AsSupportsStorageCredentials()) {
70+
// First, so the FileIO never briefly holds credentials it cannot replace.
71+
if (refresher) {
72+
credentialed->SetCredentialRefresher(std::move(refresher));
73+
}
6874
ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(storage_credentials));
6975
} else {
7076
return NotSupported("Configured FileIO does not support vended storage credentials");

‎src/iceberg/catalog/rest/rest_file_io.h‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,9 +41,13 @@ ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(
4141
const RestCatalogProperties& config);
4242

4343
/// \brief Build the configured table FileIO and apply storage credentials if present.
44+
///
45+
/// \param refresher Optional callback used to replace the vended credentials
46+
/// before they expire; ignored when the FileIO cannot tell when they do.
4447
ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeTableFileIO(
4548
const std::unordered_map<std::string, std::string>& catalog_config,
4649
const std::unordered_map<std::string, std::string>& table_config,
47-
const std::vector<StorageCredential>& storage_credentials);
50+
const std::vector<StorageCredential>& storage_credentials,
51+
StorageCredentialRefresher refresher = nullptr);
4852

4953
} // namespace iceberg::rest

‎src/iceberg/catalog/rest/types.h‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -209,6 +209,21 @@ using CreateTableResponse = LoadTableResult;
209209
/// \brief Alias of LoadTableResult used as the body of LoadTableResponse
210210
using LoadTableResponse = LoadTableResult;
211211

212+
/// \brief Response body of the LoadCredentials API.
213+
struct ICEBERG_REST_EXPORT LoadCredentialsResponse {
214+
std::vector<StorageCredential> storage_credentials;
215+
216+
/// \brief Validates the LoadCredentialsResponse.
217+
Status Validate() const {
218+
for (const auto& credential : storage_credentials) {
219+
ICEBERG_RETURN_UNEXPECTED(credential.Validate());
220+
}
221+
return {};
222+
}
223+
224+
bool operator==(const LoadCredentialsResponse& other) const = default;
225+
};
226+
212227
/// \brief Response body for listing namespaces.
213228
struct ICEBERG_REST_EXPORT ListNamespacesResponse {
214229
PageToken next_page_token;

‎src/iceberg/file_io.h‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,11 @@ class ICEBERG_EXPORT SupportsStorageCredentials {
198198
/// By value because a concurrent install may replace them. An implementation
199199
/// that delegates may report what was installed on it.
200200
virtual std::vector<StorageCredential> credentials() const = 0;
201+
202+
/// \brief Install a callback that re-fetches credentials before they expire.
203+
///
204+
/// Ignored by implementations that cannot tell when theirs expire.
205+
virtual void SetCredentialRefresher(StorageCredentialRefresher /*refresher*/) {}
201206
};
202207

203208
} // namespace iceberg

‎src/iceberg/resolving_file_io.cc‎

Lines changed: 29 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -53,8 +53,12 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
5353
ICEBERG_ASSIGN_OR_RAISE(auto io, FileIORegistry::Load(name, properties_));
5454
// Forward all credentials; each implementation applies the prefixes it
5555
// understands.
56-
if (!storage_credentials_.empty()) {
57-
if (auto* credentialed = io->AsSupportsStorageCredentials()) {
56+
if (auto* credentialed = io->AsSupportsStorageCredentials()) {
57+
// Before the credentials, so the delegate can always replace them.
58+
if (refresher_) {
59+
credentialed->SetCredentialRefresher(refresher_);
60+
}
61+
if (!storage_credentials_.empty()) {
5862
ICEBERG_RETURN_UNEXPECTED(
5963
credentialed->SetStorageCredentials(storage_credentials_));
6064
}
@@ -103,9 +107,13 @@ Status ResolvingFileIO::SetStorageCredentials(
103107
const std::vector<StorageCredential>& storage_credentials) {
104108
// Rebuild delegates lazily with the new credentials. Updating live delegates
105109
// instead would leave the resolver inconsistent if one of them rejected them.
106-
std::unique_lock lock(mutex_);
107-
storage_credentials_ = storage_credentials;
108-
io_by_name_.clear();
110+
// Retired outside the lock: tearing down a delegate can block.
111+
decltype(io_by_name_) retired;
112+
{
113+
std::unique_lock lock(mutex_);
114+
storage_credentials_ = storage_credentials;
115+
retired.swap(io_by_name_);
116+
}
109117
return {};
110118
}
111119

@@ -114,4 +122,20 @@ std::vector<StorageCredential> ResolvingFileIO::credentials() const {
114122
return storage_credentials_;
115123
}
116124

125+
void ResolvingFileIO::SetCredentialRefresher(StorageCredentialRefresher refresher) {
126+
// Drop the cached delegates so they are rebuilt with the refresher. Retired
127+
// outside the lock: teardown can block, and the outgoing callback's captures
128+
// must not destruct under `mutex_`.
129+
decltype(io_by_name_) retired;
130+
// Holds the incoming callback going in and the outgoing one coming out; a
131+
// pure swap never destroys a target under the lock, which std::exchange's
132+
// move is permitted to do.
133+
StorageCredentialRefresher handoff = std::move(refresher);
134+
{
135+
std::unique_lock lock(mutex_);
136+
refresher_.swap(handoff);
137+
retired.swap(io_by_name_);
138+
}
139+
}
140+
117141
} // namespace iceberg

‎src/iceberg/resolving_file_io.h‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,8 +39,8 @@ namespace iceberg {
3939

4040
/// \brief FileIO that resolves and caches implementations by registry name.
4141
///
42-
/// Vended credentials are forwarded to every resolved implementation that
43-
/// supports them; each applies what it understands.
42+
/// Vended credentials and any refresher are forwarded to every resolved
43+
/// implementation that supports them; each applies what it understands.
4444
class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
4545
public SupportsStorageCredentials {
4646
public:
@@ -61,8 +61,18 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
6161
Status SetStorageCredentials(
6262
const std::vector<StorageCredential>& storage_credentials) override;
6363

64+
/// \brief Return the credentials installed on this resolver.
65+
///
66+
/// Not necessarily the ones in use: a resolved implementation refreshes its
67+
/// own without reporting back.
6468
std::vector<StorageCredential> credentials() const override;
6569

70+
/// \brief Install a refresher and rebuild cached delegates with it.
71+
///
72+
/// A delegate dropped here is rebuilt from the credentials installed on this
73+
/// resolver, which may be older than what it had refreshed itself to.
74+
void SetCredentialRefresher(StorageCredentialRefresher refresher) override;
75+
6676
SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
6777

6878
private:
@@ -73,6 +83,7 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
7383
// Guards lazy resolution and credential state.
7484
mutable std::shared_mutex mutex_;
7585
std::vector<StorageCredential> storage_credentials_;
86+
StorageCredentialRefresher refresher_;
7687
std::unordered_map<std::string, std::shared_ptr<FileIO>, StringHash, StringEqual>
7788
io_by_name_;
7889
};

0 commit comments

Comments
 (0)