datalake: add iceberg_rest_catalog_credentials_host config

When cloud_storage_credentials_host is set (e.g. to an STS endpoint),
it was being applied to Iceberg REST catalog credential requests too,
regardless of the catalog's credentials source. This caused failures
when the catalog uses aws_instance_metadata (which expects
169.254.169.254) but the cloud storage host override points elsewhere.

Add iceberg_rest_catalog_credentials_host to allow the catalog
credentials host to be set independently. Thread host_override through
build_refresh_credentials_source and s3_compat_config down to
make_refresh_credentials, replacing the direct config read in the
template with an explicit parameter.

Author: wdberkeley

src/v/cloud_roles/auth_refresh_bg_op.cc

@@ -57,6 +57,7 @@ void auth_refresh_bg_op::do_start_auth_refresh_op(
         try {
             cloud_roles::aws_service_name service_name;
             cloud_roles::aws_region_name region_name;
+            std::optional<ss::sstring> host_override;
 
             ss::visit(
               _source_config,
@@ -69,6 +70,7 @@ void auth_refresh_bg_op::do_start_auth_refresh_op(
                   s3_cfg) {
                   service_name = s3_cfg.service;
                   region_name = s3_cfg.region;
+                  host_override = s3_cfg.host;
               },
               [&](const cloud_roles::auth_refresh_bg_op::abs_config&) {});
 
@@ -80,6 +82,7 @@ void auth_refresh_bg_op::do_start_auth_refresh_op(
                 service_name,
                 region_name,
                 std::nullopt,
+                std::move(host_override),
                 cloud_roles::default_retry_params,
                 std::move(metrics_tag)));
 

src/v/cloud_roles/auth_refresh_bg_op.h

@@ -29,6 +29,7 @@ class auth_refresh_bg_op {
     struct s3_compat_config {
         cloud_roles::aws_service_name service;
         cloud_roles::aws_region_name region;
+        std::optional<ss::sstring> host;
     };
     struct abs_config {};
 

src/v/cloud_roles/refresh_credentials.cc

@@ -406,6 +406,7 @@ refresh_credentials make_refresh_credentials(
   aws_service_name service,
   aws_region_name region,
   std::optional<net::unresolved_address> endpoint,
+  std::optional<ss::sstring> host_override,
   retry_params retry_params,
   ss::sstring metrics_tag) {
     switch (cloud_credentials_source) {
@@ -423,6 +424,7 @@ refresh_credentials make_refresh_credentials(
           std::move(service),
           std::move(region),
           std::move(endpoint),
+          std::move(host_override),
           retry_params,
           std::move(metrics_tag));
     case model::cloud_credentials_source::sts:
@@ -432,6 +434,7 @@ refresh_credentials make_refresh_credentials(
           std::move(service),
           std::move(region),
           std::move(endpoint),
+          std::move(host_override),
           retry_params,
           std::move(metrics_tag));
     case model::cloud_credentials_source::gcp_instance_metadata:
@@ -441,6 +444,7 @@ refresh_credentials make_refresh_credentials(
           std::move(service),
           std::move(region),
           std::move(endpoint),
+          std::move(host_override),
           retry_params,
           std::move(metrics_tag));
     case model::cloud_credentials_source::azure_aks_oidc_federation:
@@ -450,6 +454,7 @@ refresh_credentials make_refresh_credentials(
           std::move(service),
           std::move(region),
           std::move(endpoint),
+          std::move(host_override),
           retry_params,
           std::move(metrics_tag));
     case model::cloud_credentials_source::azure_vm_instance_metadata:
@@ -459,6 +464,7 @@ refresh_credentials make_refresh_credentials(
           std::move(service),
           std::move(region),
           std::move(endpoint),
+          std::move(host_override),
           retry_params,
           std::move(metrics_tag));
     }

src/v/cloud_roles/refresh_credentials.h

@@ -188,6 +188,7 @@ refresh_credentials make_refresh_credentials(
   aws_service_name service,
   aws_region_name region,
   std::optional<net::unresolved_address> endpoint = std::nullopt,
+  std::optional<ss::sstring> host_override = std::nullopt,
   retry_params retry_params = default_retry_params,
   ss::sstring metrics_tag = "") {
     ss::sstring host = {
@@ -196,17 +197,14 @@ refresh_credentials make_refresh_credentials(
     if (endpoint) {
         host = endpoint->host();
     }
-    if (
-      auto cfg_host
-      = config::shard_local_cfg().cloud_storage_credentials_host();
-      cfg_host.has_value()) {
+    if (host_override.has_value()) {
         vlog(
           clrl_log.info,
           "overriding default cloud roles credentials host {} with {} set "
           "in configuration.",
           host,
-          cfg_host.value());
-        host = cfg_host.value();
+          host_override.value());
+        host = host_override.value();
     }
     auto port = endpoint ? endpoint->port() : CredentialsProvider::default_port;
     auto impl = std::make_unique<CredentialsProvider>(
@@ -232,6 +230,7 @@ refresh_credentials make_refresh_credentials(
   aws_service_name service,
   aws_region_name region,
   std::optional<net::unresolved_address> endpoint = std::nullopt,
+  std::optional<ss::sstring> host_override = std::nullopt,
   retry_params retry_params = default_retry_params,
   ss::sstring metrics_tag = "");
 

src/v/cloud_storage_clients/configuration.cc

@@ -342,7 +342,8 @@ std::ostream& operator<<(std::ostream& o, const client_configuration& c) {
 cloud_roles::auth_refresh_bg_op::credentials_source_config
 build_refresh_credentials_source(
   const client_configuration& config,
-  model::cloud_credentials_source cloud_credentials_source) {
+  model::cloud_credentials_source cloud_credentials_source,
+  std::optional<ss::sstring> host_override) {
     if (
       cloud_credentials_source
       == model::cloud_credentials_source::config_file) {
@@ -366,10 +367,12 @@ build_refresh_credentials_source(
     } else {
         return ss::visit(
           config,
-          [](const cloud_storage_clients::s3_configuration& s3_cfg)
+          [&](const cloud_storage_clients::s3_configuration& s3_cfg)
             -> cloud_roles::auth_refresh_bg_op::credentials_source_config {
               return cloud_roles::auth_refresh_bg_op::s3_compat_config{
-                .service = s3_cfg.service, .region = s3_cfg.region};
+                .service = s3_cfg.service,
+                .region = s3_cfg.region,
+                .host = host_override};
           },
           [](const cloud_storage_clients::abs_configuration&)
             -> cloud_roles::auth_refresh_bg_op::credentials_source_config {

src/v/cloud_storage_clients/configuration.h

@@ -239,7 +239,8 @@ model::cloud_storage_backend infer_backend_from_configuration(
 cloud_roles::auth_refresh_bg_op::credentials_source_config
 build_refresh_credentials_source(
   const client_configuration& config,
-  model::cloud_credentials_source cloud_credentials_source);
+  model::cloud_credentials_source cloud_credentials_source,
+  std::optional<ss::sstring> host_override = std::nullopt);
 
 ss::future<ss::shared_ptr<ss::tls::certificate_credentials>>
 build_tls_credentials(const client_configuration& config);

src/v/cloud_storage_clients/credential_manager.cc

@@ -21,7 +21,7 @@ credential_manager::credential_manager(
   model::cloud_credentials_source cloud_credentials_source)
   : _upstream(upstream)
   , _client_conf(std::move(conf))
-  , _auth_refresh_bg_op{pool_log, _gate, _as, cloud_credentials_source, cloud_storage_clients::build_refresh_credentials_source(_client_conf, cloud_credentials_source)}
+  , _auth_refresh_bg_op{pool_log, _gate, _as, cloud_credentials_source, cloud_storage_clients::build_refresh_credentials_source(_client_conf, cloud_credentials_source, config::shard_local_cfg().cloud_storage_credentials_host())}
   , _azure_shared_key_binding(
       config::shard_local_cfg().cloud_storage_azure_shared_key.bind()) {}
 

src/v/config/configuration.cc

@@ -4316,6 +4316,21 @@ configuration::configuration()
       {.needs_restart = needs_restart::yes, .visibility = visibility::user},
       std::nullopt,
       &validate_non_empty_string_opt)
+  , iceberg_rest_catalog_credentials_host(
+      *this,
+      "iceberg_rest_catalog_credentials_host",
+      "The hostname to connect to for retrieving role based credentials for "
+      "the Iceberg REST catalog. May be required when the REST catalog uses a "
+      "different credentials source than cloud storage -- for example, if "
+      "cloud_storage_credentials_source is set to sts with a custom "
+      "credentials "
+      "host, but the REST catalog uses aws_instance_metadata, this should be "
+      "set to the EC2 instance metadata address (169.254.169.254) to prevent "
+      "the cloud storage host override from applying to catalog credential "
+      "requests.",
+      {.needs_restart = needs_restart::yes, .visibility = visibility::user},
+      std::nullopt,
+      &validate_non_empty_string_opt)
   , iceberg_backlog_controller_p_coeff(
       *this,
       "iceberg_backlog_controller_p_coeff",

src/v/config/configuration.h

@@ -743,6 +743,7 @@ struct configuration final : public config_store {
     enum_property<std::optional<model::cloud_credentials_source>>
       iceberg_rest_catalog_aws_credentials_source;
     property<std::optional<ss::sstring>> iceberg_rest_catalog_gcp_user_project;
+    property<std::optional<ss::sstring>> iceberg_rest_catalog_credentials_host;
     property<double> iceberg_backlog_controller_p_coeff;
     property<double> iceberg_backlog_controller_i_coeff;
     bounded_property<uint32_t> iceberg_target_backlog_size;

src/v/datalake/credential_manager.cc

@@ -235,7 +235,9 @@ void credential_manager::start_auth_refresh_if_needed() {
 
     auto config_source
       = cloud_storage_clients::build_refresh_credentials_source(
-        *client_config, get_credentials_source(cfg));
+        *client_config,
+        get_credentials_source(cfg),
+        cfg.iceberg_rest_catalog_credentials_host());
 
     auth_refresh_bg_op_.emplace(
       datalake_log,