cloud_storage: remote labels

src/v/cloud_storage/remote.cc

@@ -231,33 +231,6 @@ ss::future<download_result> remote::maybe_download_manifest(
     co_return co_await do_download_manifest(bucket, fk, manifest, parent, true);
 }
 
-ss::future<std::pair<download_result, manifest_format>>
-remote::try_download_partition_manifest(
-  const cloud_storage_clients::bucket_name& bucket,
-  partition_manifest& manifest,
-  retry_chain_node& parent,
-  bool expect_missing) {
-    vassert(
-      manifest.get_ntp() != model::ntp{}
-        && manifest.get_revision_id() != model::initial_revision_id{},
-      "partition manifest must have ntp");
-
-    // first try to download the serde format
-    auto format_path = manifest.get_manifest_format_and_path();
-    auto serde_result = co_await do_download_manifest(
-      bucket, format_path, manifest, parent, expect_missing);
-    if (serde_result != download_result::notfound) {
-        // propagate success, timedout and failed to caller
-        co_return std::pair{serde_result, manifest_format::serde};
-    }
-    // fallback to json format
-    format_path = manifest.get_legacy_manifest_format_and_path();
-    co_return std::pair{
-      co_await do_download_manifest(
-        bucket, format_path, manifest, parent, expect_missing),
-      manifest_format::json};
-}
-
 ss::future<download_result> remote::do_download_manifest(
   const cloud_storage_clients::bucket_name& bucket,
   const std::pair<manifest_format, remote_manifest_path>& format_key,

src/v/cloud_storage/remote.h

@@ -268,18 +268,6 @@ class remote
       base_manifest& manifest,
       retry_chain_node& parent);
 
-    /// \brief Try downloading partition_manifest. the function tries first the
-    /// manifest_format::serde path, and then manifest_format::json path. it's
-    /// expected that manifest is constructed with the approprieate npt and
-    /// revision_id, as it will be used to generate the paths return type is
-    /// download_result and index of path that generated the result
-    ss::future<std::pair<download_result, manifest_format>>
-    try_download_partition_manifest(
-      const cloud_storage_clients::bucket_name& bucket,
-      partition_manifest& manifest,
-      retry_chain_node& parent,
-      bool expect_missing = false);
-
     /// \brief Upload manifest to the pre-defined S3 location
     ///
     /// \param bucket is a bucket name

src/v/cloud_storage/tests/remote_test.cc

@@ -16,6 +16,7 @@
 #include "cloud_storage/materialized_resources.h"
 #include "cloud_storage/offset_translation_layer.h"
 #include "cloud_storage/partition_manifest.h"
+#include "cloud_storage/partition_manifest_downloader.h"
 #include "cloud_storage/remote.h"
 #include "cloud_storage/remote_segment.h"
 #include "cloud_storage/tests/common_def.h"
@@ -88,15 +89,6 @@ static cloud_storage::lazy_abort_source always_continue{
 static constexpr model::cloud_credentials_source config_file{
   model::cloud_credentials_source::config_file};
 
-static partition_manifest load_manifest_from_str(std::string_view v) {
-    partition_manifest m;
-    iobuf i;
-    i.append(v.data(), v.size());
-    auto s = make_iobuf_input_stream(std::move(i));
-    m.update(manifest_format::json, std::move(s)).get();
-    return m;
-}
-
 static remote::event_filter allow_all;
 
 static iobuf make_iobuf_from_string(std::string_view s) {
@@ -154,24 +146,6 @@ using remote_fixture = remote_fixture_base<noop_mixin_t>;
 using gcs_remote_fixture = remote_fixture_base<
   backend_override_mixin_t<model::cloud_storage_backend::google_s3_compat>>;
 
-static auto run_manifest_download_and_check(
-  auto& remote,
-  const cloud_storage_clients::bucket_name& bucket_name,
-  partition_manifest expected_manifest,
-  manifest_format expected_download_format,
-  std::string_view test_context) {
-    partition_manifest actual(manifest_ntp, manifest_revision);
-    retry_chain_node fib(never_abort, 100ms, 20ms);
-    auto [res, fmt] = remote.local()
-                        .try_download_partition_manifest(
-                          bucket_name, actual, fib)
-                        .get();
-
-    EXPECT_TRUE(res == download_result::success);
-    EXPECT_TRUE(fmt == expected_download_format);
-    EXPECT_TRUE(expected_manifest == actual);
-}
-
 class all_types_remote_fixture
   : public remote_fixture
   , public testing::TestWithParam<remote_test_parameters> {
@@ -188,44 +162,6 @@ class all_types_gcs_remote_fixture
       : gcs_remote_fixture(GetParam().url_style) {}
 };
 
-TEST_P(all_types_remote_fixture, test_download_manifest_json) {
-    set_expectations_and_listen({expectation{
-      .url = manifest_url, .body = ss::sstring(manifest_payload)}});
-    auto subscription = remote.local().subscribe(allow_all);
-    run_manifest_download_and_check(
-      remote,
-      bucket_name,
-      load_manifest_from_str(manifest_payload),
-      manifest_format::json,
-      "manifest load from json");
-    EXPECT_TRUE(subscription.available());
-    EXPECT_TRUE(
-      subscription.get().type == api_activity_type::manifest_download);
-}
-
-TEST_P(all_types_remote_fixture, test_download_manifest_serde) {
-    auto translator = load_manifest_from_str(manifest_payload);
-    auto serialized = translator.serialize().get();
-    auto manifest_binary
-      = serialized.stream.read_exactly(serialized.size_bytes).get();
-
-    set_expectations_and_listen({expectation{
-      .url = manifest_serde_url,
-      .body = ss::sstring{manifest_binary.begin(), manifest_binary.end()}}});
-
-    auto subscription = remote.local().subscribe(allow_all);
-    run_manifest_download_and_check(
-      remote,
-      bucket_name,
-      std::move(translator),
-      manifest_format::serde,
-      "manifest load from serde");
-
-    EXPECT_TRUE(subscription.available());
-    EXPECT_TRUE(
-      subscription.get().type == api_activity_type::manifest_download);
-}
-
 TEST_P(all_types_remote_fixture, test_download_manifest_timeout) { // NOLINT
     partition_manifest actual(manifest_ntp, manifest_revision);
     auto subscription = remote.local().subscribe(allow_all);
@@ -1378,8 +1314,15 @@ TEST_P(all_types_remote_fixture, test_notification_retry_meta) {
     partition_manifest actual(manifest_ntp, manifest_revision);
     auto filter = remote::event_filter{};
 
-    auto fut = remote.local().try_download_partition_manifest(
-      bucket_name, actual, fib);
+    remote_path_provider path_provider(std::nullopt);
+    partition_manifest_downloader dl(
+      bucket_name,
+      path_provider,
+      manifest_ntp,
+      manifest_revision,
+      remote.local());
+
+    auto fut = dl.download_manifest(fib, &actual);
 
     RPTEST_REQUIRE_EVENTUALLY(2s, [&] {
         auto sub = remote.local().subscribe(filter);
@@ -1388,8 +1331,9 @@ TEST_P(all_types_remote_fixture, test_notification_retry_meta) {
         });
     });
 
-    auto [res, fmt] = fut.get();
-    EXPECT_TRUE(res == download_result::timedout);
+    auto res = fut.get();
+    EXPECT_TRUE(res.has_error());
+    EXPECT_TRUE(res.error() == error_outcome::manifest_download_error);
 }
 
 TEST_P(all_types_remote_fixture, test_get_object) {

src/v/cloud_storage/tests/util.cc

@@ -10,6 +10,7 @@
  */
 #include "cloud_storage/tests/util.h"
 
+#include "cloud_storage/partition_manifest_downloader.h"
 #include "model/record.h"
 #include "model/record_batch_types.h"
 
@@ -621,11 +622,15 @@ partition_manifest hydrate_manifest(
   remote& api, const cloud_storage_clients::bucket_name& bucket) {
     static ss::abort_source never_abort;
 
+    remote_path_provider path_provider(std::nullopt);
+    partition_manifest_downloader dl(
+      bucket, path_provider, manifest_ntp, manifest_revision, api);
     partition_manifest m(manifest_ntp, manifest_revision);
-    ss::lowres_clock::update();
     retry_chain_node rtc(never_abort, 300s, 200ms);
-    auto [res, _] = api.try_download_partition_manifest(bucket, m, rtc).get();
-    BOOST_REQUIRE(res == cloud_storage::download_result::success);
+    ss::lowres_clock::update();
+    auto res = dl.download_manifest(rtc, &m).get();
+    BOOST_REQUIRE(res.has_value());
+    BOOST_REQUIRE(res.value() == find_partition_manifest_outcome::success);
     return m;
 }