cloud_storage/tests: use better containers in s3_imposter

Saw an instance where we failed to allocate ~800KB because one of the
containers in s3_imposters grew to be large. It isn't unreasonable to
expect a test to bombard s3_imposter with requests, especially when
trying to stress certain subsystems.

This replaces the containers in s3_imposter with chunked_vector and
absl::btree_multimap, which are more aligned with best practices of our
codebase.

Author: andrwng

src/v/cloud_io/tests/BUILD

@@ -34,9 +34,11 @@ redpanda_test_cc_library(
         "//src/v/base",
         "//src/v/cloud_storage_clients",
         "//src/v/config",
+        "//src/v/container:chunked_vector",
         "//src/v/http:utils",
         "//src/v/http/tests:utils",
         "//src/v/utils:uuid",
+        "@abseil-cpp//absl/container:btree",
         "@abseil-cpp//absl/container:flat_hash_set",
         "@seastar",
     ],

src/v/cloud_io/tests/s3_imposter.cc

@@ -39,7 +39,7 @@ uint16_t unit_test_httpd_port_number() { return 4442; }
 namespace {
 
 using expectation_map_t
-  = std::map<ss::sstring, s3_imposter_fixture::expectation>;
+  = absl::btree_map<ss::sstring, s3_imposter_fixture::expectation>;
 
 // Takes the input map of keys to expectations and returns a stringified XML
 // corresponding to the appropriate S3 response.
@@ -49,7 +49,7 @@ ss::sstring list_objects_resp(
   ss::sstring delimiter,
   std::optional<size_t> max_keys_opt,
   std::optional<ss::sstring> continuation_token_opt) {
-    std::map<ss::sstring, size_t> content_key_to_size;
+    absl::btree_map<ss::sstring, size_t> content_key_to_size;
     std::set<ss::sstring> common_prefixes;
     // Filter by prefix and group by the substring between the prefix and first
     // delimiter.
@@ -165,7 +165,7 @@ uint64_t string_view_to_ul(std::string_view sv) {
 
 struct s3_imposter_fixture::content_handler {
     content_handler(
-      const std::vector<s3_imposter_fixture::expectation>& exp,
+      const chunked_vector<s3_imposter_fixture::expectation>& exp,
       s3_imposter_fixture& imp,
       std::optional<absl::flat_hash_set<ss::sstring>> headers_to_store
       = std::nullopt)
@@ -176,13 +176,13 @@ struct s3_imposter_fixture::content_handler {
         }
     }
 
-    void insert(const std::vector<s3_imposter_fixture::expectation>& list) {
+    void insert(const chunked_vector<s3_imposter_fixture::expectation>& list) {
         for (const auto& exp : list) {
             expectations.insert(std::make_pair(exp.url, exp));
         }
     }
 
-    void remove(const std::vector<ss::sstring>& urls) {
+    void remove(const chunked_vector<ss::sstring>& urls) {
         for (const auto& u : urls) {
             expectations.erase(u);
         }
@@ -474,15 +474,14 @@ uint16_t s3_imposter_fixture::httpd_port_number() {
     return unit_test_httpd_port_number();
 }
 
-const std::vector<http_test_utils::request_info>&
+const chunked_vector<http_test_utils::request_info>&
 s3_imposter_fixture::get_requests() const {
     return _requests;
 }
 
-std::vector<http_test_utils::request_info> s3_imposter_fixture::get_requests(
+chunked_vector<http_test_utils::request_info> s3_imposter_fixture::get_requests(
   s3_imposter_fixture::req_pred_t predicate) const {
-    std::vector<http_test_utils::request_info> matching_requests;
-    matching_requests.reserve(_requests.size());
+    chunked_vector<http_test_utils::request_info> matching_requests;
     std::copy_if(
       _requests.cbegin(),
       _requests.cend(),
@@ -491,13 +490,13 @@ std::vector<http_test_utils::request_info> s3_imposter_fixture::get_requests(
     return matching_requests;
 }
 
-const std::multimap<ss::sstring, http_test_utils::request_info>&
+const absl::btree_multimap<ss::sstring, http_test_utils::request_info>&
 s3_imposter_fixture::get_targets() const {
     return _targets;
 }
 
 void s3_imposter_fixture::set_expectations_and_listen(
-  std::vector<s3_imposter_fixture::expectation> expectations,
+  chunked_vector<s3_imposter_fixture::expectation> expectations,
   std::optional<absl::flat_hash_set<ss::sstring>> headers_to_store,
   std::set<ss::sstring> content_type_overrides) {
     const ss::sstring url_prefix = "/" + url_base();
@@ -522,7 +521,7 @@ void s3_imposter_fixture::set_expectations_and_listen(
 }
 
 void s3_imposter_fixture::add_expectations(
-  std::vector<s3_imposter_fixture::expectation> expectations) {
+  chunked_vector<s3_imposter_fixture::expectation> expectations) {
     vassert(_content_handler != nullptr, "Imposter is not initialized");
     const ss::sstring url_prefix = "/" + url_base();
     for (auto& expectation : expectations) {
@@ -532,7 +531,8 @@ void s3_imposter_fixture::add_expectations(
     _content_handler->insert(expectations);
 }
 
-void s3_imposter_fixture::remove_expectations(std::vector<ss::sstring> urls) {
+void s3_imposter_fixture::remove_expectations(
+  chunked_vector<ss::sstring> urls) {
     vassert(_content_handler != nullptr, "Imposter is not initialized");
     const ss::sstring url_prefix = "/" + url_base();
     for (auto& url : urls) {
@@ -561,7 +561,7 @@ ss::sstring s3_imposter_fixture::url_base() const {
 
 void s3_imposter_fixture::set_routes(
   ss::httpd::routes& r,
-  const std::vector<s3_imposter_fixture::expectation>& expectations,
+  const chunked_vector<s3_imposter_fixture::expectation>& expectations,
   std::optional<absl::flat_hash_set<ss::sstring>> headers_to_store,
   std::set<ss::sstring> content_type_overrides) {
     using namespace ss::httpd;
@@ -626,9 +626,9 @@ cloud_storage_clients::http_byte_range parse_byte_header(std::string_view s) {
       string_view_to_ul(bytes_value.substr(split_at + 1)));
 }
 
-std::vector<cloud_storage_clients::object_key>
+chunked_vector<cloud_storage_clients::object_key>
 keys_from_delete_objects_request(const http_test_utils::request_info& req) {
-    std::vector<cloud_storage_clients::object_key> keys;
+    chunked_vector<cloud_storage_clients::object_key> keys;
 
     auto buffer_stream = std::istringstream{std::string{req.content}};
     boost::property_tree::ptree tree;
@@ -643,9 +643,10 @@ keys_from_delete_objects_request(const http_test_utils::request_info& req) {
     return keys;
 }
 
-std::vector<std::pair<ss::sstring, cloud_storage_clients::object_key>>
+chunked_vector<std::pair<ss::sstring, cloud_storage_clients::object_key>>
 keys_from_batch_delete_request(const http_test_utils::request_info& req) {
-    std::vector<std::pair<ss::sstring, cloud_storage_clients::object_key>> keys;
+    chunked_vector<std::pair<ss::sstring, cloud_storage_clients::object_key>>
+      keys;
     auto buffer_stream = std::istringstream{std::string{req.content}};
 
     // crudely iterate over request lines, stripping out object keys

src/v/cloud_io/tests/s3_imposter.h

@@ -10,10 +10,12 @@
 
 #pragma once
 
+#include "absl/container/btree_map.h"
 #include "absl/container/flat_hash_set.h"
 #include "base/seastarx.h"
 #include "cloud_storage_clients/client.h"
 #include "config/configuration.h"
+#include "container/chunked_vector.h"
 #include "http/tests/registered_urls.h"
 #include "utils/uuid.h"
 
@@ -22,9 +24,6 @@
 #include <seastar/core/sstring.hh>
 #include <seastar/http/httpd.hh>
 
-#include <map>
-#include <vector>
-
 inline cloud_storage_clients::bucket_name random_test_bucket_name() {
     return cloud_storage_clients::bucket_name{
       "test-bucket-" + ss::sstring{uuid_t::create()}};
@@ -81,30 +80,30 @@ class s3_imposter_fixture {
     /// to null but there was PUT call that sent some data, subsequent GET call
     /// will retrieve this data.
     void set_expectations_and_listen(
-      std::vector<expectation> expectations,
+      chunked_vector<expectation> expectations,
       std::optional<absl::flat_hash_set<ss::sstring>> headers_to_store
       = std::nullopt,
       std::set<ss::sstring> content_type_overrides = {});
 
     /// Update expectations for the REST API.
-    void add_expectations(std::vector<expectation> expectations);
-    void remove_expectations(std::vector<ss::sstring> urls);
+    void add_expectations(chunked_vector<expectation> expectations);
+    void remove_expectations(chunked_vector<ss::sstring> urls);
 
     /// Get object from S3 or nullopt if it doesn't exist
     std::optional<ss::sstring> get_object(const ss::sstring& url) const;
 
     /// Access all http requests ordered by time
-    const std::vector<http_test_utils::request_info>& get_requests() const;
+    const chunked_vector<http_test_utils::request_info>& get_requests() const;
 
     using req_pred_t
       = std::function<bool(const http_test_utils::request_info&)>;
 
     /// Access http requests matching the given predicate
-    std::vector<http_test_utils::request_info>
+    chunked_vector<http_test_utils::request_info>
     get_requests(req_pred_t predicate) const;
 
     /// Access all http requests ordered by target url
-    const std::multimap<ss::sstring, http_test_utils::request_info>&
+    const absl::btree_multimap<ss::sstring, http_test_utils::request_info>&
     get_targets() const;
 
     cloud_storage_clients::s3_configuration get_configuration();
@@ -127,7 +126,7 @@ class s3_imposter_fixture {
 private:
     void set_routes(
       ss::httpd::routes& r,
-      const std::vector<expectation>& expectations,
+      const chunked_vector<expectation>& expectations,
       std::optional<absl::flat_hash_set<ss::sstring>> headers_to_store
       = std::nullopt,
       std::set<ss::sstring> content_type_overrides = {});
@@ -140,16 +139,16 @@ class s3_imposter_fixture {
     ss::shared_ptr<content_handler> _content_handler;
     std::unique_ptr<ss::httpd::handler_base> _handler;
     /// Contains saved requests
-    std::vector<http_test_utils::request_info> _requests;
+    chunked_vector<http_test_utils::request_info> _requests;
     /// Contains all accessed target urls
-    std::multimap<ss::sstring, http_test_utils::request_info> _targets;
+    absl::btree_multimap<ss::sstring, http_test_utils::request_info> _targets;
 
     /// Whether or not to search through expectations for content when handling
     /// a list GET request.
     bool _search_on_get_list{true};
 
-    std::vector<req_pred_t> _fail_request_if;
-    std::vector<http_test_utils::response> _failure_response;
+    chunked_vector<req_pred_t> _fail_request_if;
+    chunked_vector<http_test_utils::response> _failure_response;
 };
 
 class enable_cloud_storage_fixture {
@@ -163,8 +162,8 @@ class enable_cloud_storage_fixture {
 
 cloud_storage_clients::http_byte_range parse_byte_header(std::string_view s);
 
-std::vector<cloud_storage_clients::object_key>
+chunked_vector<cloud_storage_clients::object_key>
 keys_from_delete_objects_request(const http_test_utils::request_info&);
 
-std::vector<std::pair<ss::sstring, cloud_storage_clients::object_key>>
+chunked_vector<std::pair<ss::sstring, cloud_storage_clients::object_key>>
 keys_from_batch_delete_request(const http_test_utils::request_info&);

src/v/cloud_storage/tests/BUILD

@@ -43,6 +43,7 @@ redpanda_test_cc_library(
         "//src/v/cloud_storage:types",
         "//src/v/cluster",
         "//src/v/cluster/cloud_metadata/tests:manual_mixin",
+        "//src/v/container:chunked_vector",
         "//src/v/json",
         "//src/v/kafka/server/tests:kafka_test_utils",
         "//src/v/model",

src/v/cloud_storage/tests/async_manifest_view_test.cc

@@ -244,7 +244,7 @@ class async_manifest_view_fixture
         }
     }
 
-    void listen() { set_expectations_and_listen(_expectations); }
+    void listen() { set_expectations_and_listen(std::move(_expectations)); }
 
     void collect_segments_to(std::vector<segment_meta>& meta) {
         all_segments = std::ref(meta);
@@ -330,7 +330,7 @@ class async_manifest_view_fixture
     retry_chain_logger ctxlog;
     partition_probe probe;
     async_manifest_view view;
-    std::vector<expectation> _expectations;
+    chunked_vector<expectation> _expectations;
     std::vector<model::offset> spillover_start_offsets;
     model::offset _last_spillover_offset;
     std::optional<std::reference_wrapper<std::vector<segment_meta>>>

src/v/cloud_storage/tests/remote_partition_test.cc

@@ -421,7 +421,7 @@ FIXTURE_TEST(test_overlapping_segments, cloud_storage_fixture) {
       .url = prefixed_partition_manifest_json_path(
         manifest.get_ntp(), manifest.get_revision_id()),
       .body = body};
-    set_expectations_and_listen(expectations);
+    set_expectations_and_listen(std::move(expectations));
     BOOST_REQUIRE(check_scan(*this, kafka::offset(0), 9));
 }
 
@@ -2306,7 +2306,7 @@ FIXTURE_TEST(test_out_of_range_query, cloud_storage_fixture) {
     cloud_storage::partition_manifest manifest(manifest_ntp, manifest_revision);
 
     auto expectations = make_imposter_expectations(manifest, segments);
-    set_expectations_and_listen(expectations);
+    set_expectations_and_listen(std::move(expectations));
 
     // Advance start offset as-if archiver did apply retention but didn't
     // run GC yet (the clean offset is not updated).
@@ -2326,11 +2326,11 @@ FIXTURE_TEST(test_out_of_range_query, cloud_storage_fixture) {
       ostr.str());
 
     auto manifest_url = manifest.get_manifest_path(path_provider)().string();
-    remove_expectations({manifest_url});
-    add_expectations({
-      cloud_storage_fixture::expectation{
-        .url = manifest_url, .body = serialize_manifest(manifest)},
-    });
+    remove_expectations(chunked_vector<ss::sstring>::single(manifest_url));
+    add_expectations(
+      chunked_vector<cloud_storage_fixture::expectation>::single(
+        cloud_storage_fixture::expectation{
+          .url = manifest_url, .body = serialize_manifest(manifest)}));
 
     auto base = segments[0].base_offset;
     auto max = segments[segments.size() - 1].max_offset;
@@ -2376,7 +2376,7 @@ FIXTURE_TEST(test_out_of_range_spillover_query, cloud_storage_fixture) {
     cloud_storage::partition_manifest manifest(manifest_ntp, manifest_revision);
 
     auto expectations = make_imposter_expectations(manifest, segments);
-    set_expectations_and_listen(expectations);
+    set_expectations_and_listen(std::move(expectations));
 
     for (int i = 0; i < 2; i++) {
         spillover_manifest spm(manifest_ntp, manifest_revision);
@@ -2400,10 +2400,12 @@ FIXTURE_TEST(test_out_of_range_spillover_query, cloud_storage_fixture) {
 
         auto s_data = spm.serialize().get();
         auto buf = s_data.stream.read_exactly(s_data.size_bytes).get();
-        add_expectations({cloud_storage_fixture::expectation{
-          .url = spm.get_manifest_path(path_provider)().string(),
-          .body = ss::sstring(buf.begin(), buf.end()),
-        }});
+        add_expectations(
+          chunked_vector<cloud_storage_fixture::expectation>::single(
+            cloud_storage_fixture::expectation{
+              .url = spm.get_manifest_path(path_provider)().string(),
+              .body = ss::sstring(buf.begin(), buf.end()),
+            }));
     }
 
     // Advance start offset as-if archiver did apply retention but didn't
@@ -2430,11 +2432,11 @@ FIXTURE_TEST(test_out_of_range_spillover_query, cloud_storage_fixture) {
       ostr.str());
 
     auto manifest_url = manifest.get_manifest_path(path_provider)().string();
-    remove_expectations({manifest_url});
-    add_expectations({
-      cloud_storage_fixture::expectation{
-        .url = manifest_url, .body = serialize_manifest(manifest)},
-    });
+    remove_expectations(chunked_vector<ss::sstring>::single(manifest_url));
+    add_expectations(
+      chunked_vector<cloud_storage_fixture::expectation>::single(
+        cloud_storage_fixture::expectation{
+          .url = manifest_url, .body = serialize_manifest(manifest)}));
 
     auto base = segments[0].base_offset;
     auto max = segments[segments.size() - 1].max_offset;
@@ -2517,11 +2519,11 @@ FIXTURE_TEST(test_out_of_range_spillover_query, cloud_storage_fixture) {
       manifest.get_manifest_path(path_provider),
       ostr.str());
 
-    remove_expectations({manifest_url});
-    add_expectations({
-      cloud_storage_fixture::expectation{
-        .url = manifest_url, .body = serialize_manifest(manifest)},
-    });
+    remove_expectations(chunked_vector<ss::sstring>::single(manifest_url));
+    add_expectations(
+      chunked_vector<cloud_storage_fixture::expectation>::single(
+        cloud_storage_fixture::expectation{
+          .url = manifest_url, .body = serialize_manifest(manifest)}));
 
     // Still can't query from the base offset.
     BOOST_REQUIRE_EXCEPTION(

src/v/cloud_storage/tests/remote_test.cc

@@ -835,7 +835,7 @@ TEST_P(all_types_remote_fixture, test_delete_objects_multiple_batches) {
     auto result
       = remote.local().delete_objects(bucket_name, to_delete, fib).get();
     ASSERT_EQ(cloud_storage::upload_result::success, result);
-    auto requests = get_requests();
+    const auto& requests = get_requests();
     ASSERT_EQ(requests.size(), 3);
 
     std::vector<cloud_storage_clients::object_key> deleted_keys;
@@ -875,7 +875,7 @@ TEST_P(
     auto result
       = remote.local().delete_objects(bucket_name, to_delete, fib).get();
     ASSERT_EQ(cloud_storage::upload_result::failed, result);
-    auto requests = get_requests();
+    const auto& requests = get_requests();
     ASSERT_EQ(requests.size(), 3);
 
     std::vector<cloud_storage_clients::object_key> deleted_keys;
@@ -1354,7 +1354,7 @@ TEST_P(all_types_remote_fixture, test_get_object) {
              .payload = buf})
           .get();
 
-    const auto requests = get_requests();
+    const auto& requests = get_requests();
     ASSERT_EQ(requests.size(), 2);
 
     const auto last_request = requests.back();