cloud_storage: use self configuration for url style

If the user does not specify `cloud_storage_url_style` in their
configuration files, the `s3_client` will set its url addressing style
through self configuration.

The default behavior is to attempt `virtual_host` addressing first,
falling back to `path` style requests if `virtual_host` does not work.
If both are attempted and neither work, `redpanda`'s start-up is aborted.

src/v/cloud_storage/types.cc

@@ -28,7 +28,6 @@ cloud_storage_clients::default_overrides get_default_overrides() {
         optep.has_value()) {
         overrides.endpoint = cloud_storage_clients::endpoint_url(*optep);
     }
-    overrides.url_style = config::shard_local_cfg().cloud_storage_url_style();
     overrides.disable_tls = config::shard_local_cfg().cloud_storage_disable_tls;
     if (auto cert = config::shard_local_cfg().cloud_storage_trust_file.value();
         cert.has_value()) {
@@ -412,6 +411,8 @@ ss::future<configuration> configuration::get_s3_config() {
 
     auto region = cloud_roles::aws_region_name(get_value_or_throw(
       config::shard_local_cfg().cloud_storage_region, "cloud_storage_region"));
+    auto url_style = config::shard_local_cfg().cloud_storage_url_style.value();
+
     auto disable_metrics = net::metrics_disabled(
       config::shard_local_cfg().disable_metrics());
     auto disable_public_metrics = net::public_metrics_disabled(
@@ -422,6 +423,7 @@ ss::future<configuration> configuration::get_s3_config() {
         access_key,
         secret_key,
         region,
+	url_style,
         get_default_overrides(),
         disable_metrics,
         disable_public_metrics);

src/v/cloud_storage_clients/configuration.cc

@@ -74,6 +74,7 @@ ss::future<s3_configuration> s3_configuration::make_configuration(
   const std::optional<cloud_roles::public_key_str>& pkey,
   const std::optional<cloud_roles::private_key_str>& skey,
   const cloud_roles::aws_region_name& region,
+  const std::optional<cloud_storage_clients::s3_url_style>& url_style,
   const default_overrides& overrides,
   net::metrics_disabled disable_metrics,
   net::public_metrics_disabled disable_public_metrics) {
@@ -91,7 +92,14 @@ ss::future<s3_configuration> s3_configuration::make_configuration(
     client_cfg.secret_key = skey;
     client_cfg.region = region;
     client_cfg.uri = access_point_uri(endpoint_uri);
-    client_cfg.url_style = overrides.url_style;
+
+    if (url_style.has_value()) {
+        client_cfg.url_style = url_style.value();
+    } else {
+        // If the url style is not specified, it will be determined with
+        // self configuration.
+        client_cfg.requires_self_configuration = true;
+    }
 
     if (overrides.disable_tls == false) {
         client_cfg.credentials = co_await build_tls_credentials(
@@ -211,7 +219,10 @@ void apply_self_configuration_result(
                 "result {}",
                 cfg,
                 res);
-              // No self configuration for S3 at this point
+
+              cfg.url_style
+                = std::get<s3_self_configuration_result>(res).url_style;
+
           } else if constexpr (std::is_same_v<abs_configuration, cfg_type>) {
               vassert(
                 std::holds_alternative<abs_self_configuration_result>(res),
@@ -246,8 +257,9 @@ operator<<(std::ostream& o, const abs_self_configuration_result& r) {
     return o;
 }
 
-std::ostream& operator<<(std::ostream& o, const s3_self_configuration_result&) {
-    o << "{}";
+std::ostream&
+operator<<(std::ostream& o, const s3_self_configuration_result& r) {
+    o << "{s3_url_style: " << r.url_style << "}";
     return o;
 }
 

src/v/cloud_storage_clients/configuration.h

@@ -26,7 +26,6 @@ struct default_overrides {
     std::optional<uint16_t> port = std::nullopt;
     std::optional<ca_trust_file> trust_file = std::nullopt;
     std::optional<ss::lowres_clock::duration> max_idle_time = std::nullopt;
-    s3_url_style url_style = s3_url_style::virtual_host;
     bool disable_tls = false;
 };
 
@@ -50,7 +49,7 @@ struct s3_configuration : common_configuration {
     /// AWS secret key, optional if configuration uses temporary credentials
     std::optional<cloud_roles::private_key_str> secret_key;
     /// AWS URL style, either virtual-hosted-style or path-style.
-    s3_url_style url_style;
+    s3_url_style url_style = s3_url_style::virtual_host;
 
     /// \brief opinionated configuraiton initialization
     /// Generates uri field from region, initializes credentials for the
@@ -67,6 +66,7 @@ struct s3_configuration : common_configuration {
       const std::optional<cloud_roles::public_key_str>& pkey,
       const std::optional<cloud_roles::private_key_str>& skey,
       const cloud_roles::aws_region_name& region,
+      const std::optional<cloud_storage_clients::s3_url_style>& url_style,
       const default_overrides& overrides = {},
       net::metrics_disabled disable_metrics = net::metrics_disabled::yes,
       net::public_metrics_disabled disable_public_metrics
@@ -110,7 +110,9 @@ struct abs_self_configuration_result {
     bool is_hns_enabled;
 };
 
-struct s3_self_configuration_result {};
+struct s3_self_configuration_result {
+    s3_url_style url_style;
+};
 
 using client_self_configuration_output
   = std::variant<abs_self_configuration_result, s3_self_configuration_result>;

src/v/cloud_storage_clients/s3_client.cc

@@ -16,6 +16,7 @@
 #include "cloud_storage_clients/s3_error.h"
 #include "cloud_storage_clients/util.h"
 #include "cloud_storage_clients/xml_sax_parser.h"
+#include "config/configuration.h"
 #include "hashing/secure.h"
 #include "http/client.h"
 #include "net/types.h"
@@ -561,11 +562,66 @@ s3_client::s3_client(
 
 ss::future<result<client_self_configuration_output, error_outcome>>
 s3_client::self_configure() {
+    // Test virtual host style addressing, fall back to path if necessary.
+    // If any configuration options prevent testing, addressing style will
+    // default to virtual_host.
+    // If both addressing methods fail, return an error.
+    auto result = s3_self_configuration_result{
+      .url_style = s3_url_style::virtual_host};
+    if (!config::shard_local_cfg().cloud_storage_enable_remote_read()) {
+        vlog(
+          s3_log.error,
+          "Can't self-configure S3 Client, {} is not enabled.",
+          config::shard_local_cfg().cloud_storage_enable_remote_read.name());
+        co_return result;
+    }
+
+    const auto& bucket_config = config::shard_local_cfg().cloud_storage_bucket;
+
+    if (!bucket_config.value().has_value()) {
+        vlog(
+          s3_log.error,
+          "Can't self-configure S3 Client, {} is not set.",
+          bucket_config.name());
+        co_return result;
+    }
+
+    auto bucket = cloud_storage_clients::bucket_name{
+      bucket_config.value().value()};
+
+    // Test virtual_host style.
+    vassert(
+      _requestor._ap_style == s3_url_style::virtual_host,
+      "_ap_style should be virtual host by default before self configuration "
+      "begins");
+    {
+        auto list_objects_result = co_await list_objects(
+          bucket, std::nullopt, std::nullopt, 1);
+        if (list_objects_result) {
+            // Virtual-host style request succeeded.
+            co_return result;
+        }
+    }
+
+    // Test path style.
+    _requestor._ap_style = s3_url_style::path;
+    result.url_style = _requestor._ap_style;
+    {
+        auto list_objects_result = co_await list_objects(
+          bucket, std::nullopt, std::nullopt, 1);
+        if (list_objects_result) {
+            // Path style request succeeded.
+            co_return result;
+        }
+    }
+
+    // Both addressing styles failed.
     vlog(
       s3_log.error,
-      "Call to self_configure was made, but the S3 client doesn't require self "
-      "configuration");
-    co_return s3_self_configuration_result{};
+      "Couldn't reach S3 storage with either path style or virtual_host style "
+      "requests.",
+      bucket_config.name());
+    co_return error_outcome::fail;
 }
 
 ss::future<> s3_client::stop() { return _client.stop(); }

src/v/cloud_storage_clients/s3_client.h

@@ -107,6 +107,7 @@ class request_creator {
       std::optional<char> delimiter = std::nullopt);
 
 private:
+    friend class s3_client;
     std::string make_host(const bucket_name& name) const;
 
     std::string

src/v/cloud_storage_clients/test_client/s3_test_client_main.cc

@@ -79,7 +79,7 @@ void cli_opts(boost::program_options::options_description_easy_init opt) {
 
     opt(
       "url_style",
-      po::value<std::string>()->default_value("virtual_host"),
+      po::value<std::string>()->default_value(""),
       "aws addressing style");
 
     opt(
@@ -150,11 +150,24 @@ test_conf cfg_from(boost::program_options::variables_map& m) {
     auto secret_key = cloud_roles::private_key_str(
       m["secretkey"].as<std::string>());
     auto region = cloud_roles::aws_region_name(m["region"].as<std::string>());
+    auto url_style =
+      [&]() -> std::optional<cloud_storage_clients::s3_url_style> {
+        const auto url_style_str = m["url_style"].as<std::string>();
+        if (url_style_str == "virtual_host") {
+            return cloud_storage_clients::s3_url_style::virtual_host;
+        } else if (url_style_str == "path") {
+            return cloud_storage_clients::s3_url_style::path;
+        } else {
+            return std::nullopt;
+        }
+    }();
+
     cloud_storage_clients::s3_configuration client_cfg
       = cloud_storage_clients::s3_configuration::make_configuration(
           access_key,
           secret_key,
           region,
+          url_style,
           cloud_storage_clients::default_overrides{
             .endpoint =
               [&]() -> std::optional<cloud_storage_clients::endpoint_url> {
@@ -170,13 +183,6 @@ test_conf cfg_from(boost::program_options::variables_map& m) {
                 }
                 return std::nullopt;
             }(),
-            .url_style = [&]() -> cloud_storage_clients::s3_url_style {
-                if (m["url_style"].as<std::string>() == "virtual_host") {
-                    return cloud_storage_clients::s3_url_style::virtual_host;
-                } else {
-                    return cloud_storage_clients::s3_url_style::path;
-                }
-            }(),
             .disable_tls = m.contains("disable-tls") > 0,
           })
           .get0();

src/v/config/configuration.cc

@@ -1673,14 +1673,17 @@ configuration::configuration()
   , cloud_storage_url_style(
       *this,
       "cloud_storage_url_style",
-      "The addressing style to use for S3 requests.",
+      "The addressing style to use for S3 requests. If not set, virtual_host "
+      "addressing will be the default style, falling back to path "
+      "style if required.",
       {.needs_restart = needs_restart::yes,
        .example = "virtual_host",
        .visibility = visibility::user},
-      cloud_storage_clients::s3_url_style::virtual_host,
+      std::nullopt,
       {
         cloud_storage_clients::s3_url_style::virtual_host,
         cloud_storage_clients::s3_url_style::path,
+	std::nullopt
       })
   , cloud_storage_credentials_source(
       *this,

src/v/config/configuration.h

@@ -309,7 +309,7 @@ struct configuration final : public config_store {
     property<std::optional<ss::sstring>> cloud_storage_region;
     property<std::optional<ss::sstring>> cloud_storage_bucket;
     property<std::optional<ss::sstring>> cloud_storage_api_endpoint;
-    enum_property<cloud_storage_clients::s3_url_style> cloud_storage_url_style;
+    enum_property<std::optional<cloud_storage_clients::s3_url_style>> cloud_storage_url_style;
     enum_property<model::cloud_credentials_source>
       cloud_storage_credentials_source;
     property<std::optional<ss::sstring>>