Merge pull request #11212 from andrwng/cluster-metadata-plugin

cloud_metadata: plug upload loop into application

Author: piyushredpanda

src/v/cluster/cloud_metadata/manifest_downloads.cc

@@ -22,6 +22,9 @@ namespace {
 const std::regex cluster_metadata_manifest_prefix_expr{
   R"REGEX(cluster_metadata/[a-z0-9-]+/manifests/(\d+)/)REGEX"};
 
+const std::regex cluster_metadata_manifest_expr{
+  R"REGEX(cluster_metadata/([a-z0-9-]+)/manifests/(\d+)/cluster_manifest.json)REGEX"};
+
 } // anonymous namespace
 
 namespace cluster::cloud_metadata {
@@ -72,10 +75,14 @@ ss::future<cluster_manifest_result> download_highest_manifest_for_cluster(
         if (!matches_manifest_expr) {
             continue;
         }
+        vassert(
+          matches.size() >= 2,
+          "Unexpected size of regex match",
+          matches.size());
         const auto& meta_id_str = matches[1].str();
         cluster_metadata_id meta_id;
         try {
-            meta_id = cluster_metadata_id(std::stoi(meta_id_str.c_str()));
+            meta_id = cluster_metadata_id(std::stol(meta_id_str.c_str()));
         } catch (...) {
             vlog(
               clusterlog.debug,
@@ -145,4 +152,88 @@ ss::future<std::list<ss::sstring>> list_orphaned_by_manifest(
     co_return ret;
 }
 
+ss::future<cluster_manifest_result> download_highest_manifest_in_bucket(
+  cloud_storage::remote& remote,
+  const cloud_storage_clients::bucket_name& bucket,
+  retry_chain_node& retry_node) {
+    // Look for unique cluster UUIDs for which we have metadata.
+    constexpr auto cluster_prefix = "cluster_metadata/";
+    vlog(clusterlog.trace, "Listing objects with prefix {}", cluster_prefix);
+    auto list_res = co_await remote.list_objects(
+      bucket,
+      retry_node,
+      cloud_storage_clients::object_key(cluster_prefix),
+      std::nullopt);
+    if (list_res.has_error()) {
+        vlog(clusterlog.debug, "Error downloading manifest", list_res.error());
+        co_return error_outcome::list_failed;
+    }
+    // Examine all cluster metadata in this bucket.
+    auto& cluster_metadata_items = list_res.value().contents;
+    if (cluster_metadata_items.empty()) {
+        vlog(clusterlog.debug, "No manifests found in bucket {}", bucket());
+        co_return error_outcome::no_matching_metadata;
+    }
+
+    // Look through those that look like cluster metadata manifests and find
+    // the one with the highest metadata ID. This will be the returned to the
+    // caller.
+    model::cluster_uuid uuid_with_highest_meta_id{};
+    cluster_metadata_id highest_meta_id{};
+    for (const auto& item : cluster_metadata_items) {
+        std::smatch matches;
+        std::string k = item.key;
+        const auto matches_manifest_expr = std::regex_match(
+          k.cbegin(), k.cend(), matches, cluster_metadata_manifest_expr);
+        if (!matches_manifest_expr) {
+            continue;
+        }
+        const auto& cluster_uuid_str = matches[1].str();
+        const auto& meta_id_str = matches[2].str();
+        cluster_metadata_id meta_id{};
+        model::cluster_uuid cluster_uuid{};
+        try {
+            meta_id = cluster_metadata_id(std::stoi(meta_id_str.c_str()));
+        } catch (...) {
+            vlog(
+              clusterlog.debug,
+              "Ignoring invalid metadata ID: {}",
+              meta_id_str);
+            continue;
+        }
+        try {
+            auto u = boost::lexical_cast<uuid_t::underlying_t>(
+              cluster_uuid_str);
+            std::vector<uint8_t> uuid_vec{u.begin(), u.end()};
+            cluster_uuid = model::cluster_uuid(std::move(uuid_vec));
+        } catch (...) {
+            vlog(
+              clusterlog.debug,
+              "Ignoring invalid cluster UUID: {}",
+              cluster_uuid_str);
+            continue;
+        }
+        if (meta_id > highest_meta_id) {
+            highest_meta_id = meta_id;
+            uuid_with_highest_meta_id = cluster_uuid;
+        }
+    }
+    if (highest_meta_id == cluster_metadata_id{}) {
+        vlog(clusterlog.debug, "No valid manifests in bucket {}", bucket());
+        co_return error_outcome::no_matching_metadata;
+    }
+    cluster_metadata_manifest manifest;
+    auto manifest_res = co_await remote.download_manifest(
+      bucket,
+      cluster_manifest_key(uuid_with_highest_meta_id, highest_meta_id),
+      manifest,
+      retry_node);
+    if (manifest_res != cloud_storage::download_result::success) {
+        vlog(
+          clusterlog.debug, "Manifest download failed with {}", manifest_res);
+        co_return error_outcome::download_failed;
+    }
+    co_return manifest;
+}
+
 } // namespace cluster::cloud_metadata

src/v/cluster/cloud_metadata/manifest_downloads.h

@@ -48,4 +48,15 @@ ss::future<std::list<ss::sstring>> list_orphaned_by_manifest(
   const cluster_metadata_manifest& manifest,
   retry_chain_node& retry_node);
 
+// Looks through the given bucket for cluster metadata with the highest
+// metadata ID.
+// - list_failed/download_failed: there was a physical error sending requests
+//   to remote storage, preventing us from returning an accurate result.
+// - no_matching_metadata: we were able to list an sift through the bucket, but
+//   no cluster metadata manifest exists.
+ss::future<cluster_manifest_result> download_highest_manifest_in_bucket(
+  cloud_storage::remote& remote,
+  const cloud_storage_clients::bucket_name& bucket,
+  retry_chain_node& retry_node);
+
 } // namespace cluster::cloud_metadata

src/v/cluster/cloud_metadata/tests/manifest_downloads_test.cc

@@ -95,3 +95,58 @@ FIXTURE_TEST(test_download_manifest, cluster_metadata_fixture) {
     BOOST_CHECK(m_res.has_error());
     BOOST_CHECK_EQUAL(error_outcome::list_failed, m_res.error());
 }
+
+FIXTURE_TEST(
+  test_download_highest_manifest_in_bucket, cluster_metadata_fixture) {
+    retry_chain_node retry_node(
+      never_abort, ss::lowres_clock::time_point::max(), 10ms);
+    auto m_res
+      = download_highest_manifest_in_bucket(remote, bucket, retry_node).get();
+    BOOST_REQUIRE(m_res.has_error());
+    BOOST_REQUIRE_EQUAL(m_res.error(), error_outcome::no_matching_metadata);
+
+    cluster_metadata_manifest manifest;
+    manifest.cluster_uuid = cluster_uuid;
+    manifest.metadata_id = cluster_metadata_id(10);
+    remote
+      .upload_manifest(
+        cloud_storage_clients::bucket_name("test-bucket"), manifest, retry_node)
+      .get();
+
+    m_res
+      = download_highest_manifest_in_bucket(remote, bucket, retry_node).get();
+    BOOST_REQUIRE(m_res.has_value());
+    BOOST_CHECK_EQUAL(cluster_uuid, m_res.value().cluster_uuid);
+    BOOST_CHECK_EQUAL(10, m_res.value().metadata_id());
+
+    auto new_uuid = model::cluster_uuid(uuid_t::create());
+    manifest.cluster_uuid = new_uuid;
+    manifest.metadata_id = cluster_metadata_id(15);
+
+    // Upload a new manifest with a higher metadata ID for a new cluster.
+    remote
+      .upload_manifest(
+        cloud_storage_clients::bucket_name("test-bucket"), manifest, retry_node)
+      .get();
+    m_res
+      = download_highest_manifest_in_bucket(remote, bucket, retry_node).get();
+    BOOST_REQUIRE(m_res.has_value());
+    BOOST_REQUIRE_EQUAL(15, m_res.value().metadata_id());
+    BOOST_REQUIRE_EQUAL(new_uuid, m_res.value().cluster_uuid);
+
+    // Sanity check that searching by the cluster UUIDs return the expected
+    // manifests.
+    m_res = download_highest_manifest_for_cluster(
+              remote, cluster_uuid, bucket, retry_node)
+              .get();
+    BOOST_REQUIRE(m_res.has_value());
+    BOOST_REQUIRE_EQUAL(10, m_res.value().metadata_id());
+    BOOST_REQUIRE_EQUAL(cluster_uuid, m_res.value().cluster_uuid);
+
+    m_res = download_highest_manifest_for_cluster(
+              remote, new_uuid, bucket, retry_node)
+              .get();
+    BOOST_REQUIRE(m_res.has_value());
+    BOOST_REQUIRE_EQUAL(15, m_res.value().metadata_id());
+    BOOST_REQUIRE_EQUAL(new_uuid, m_res.value().cluster_uuid);
+}

src/v/cluster/cloud_metadata/tests/uploader_test.cc

@@ -20,11 +20,13 @@
 #include "cluster/config_frontend.h"
 #include "cluster/controller_snapshot.h"
 #include "cluster/types.h"
+#include "config/configuration.h"
 #include "model/fundamental.h"
 #include "redpanda/application.h"
 #include "redpanda/tests/fixture.h"
 #include "storage/snapshot.h"
 #include "test_utils/async.h"
+#include "test_utils/scoped_config.h"
 
 #include <seastar/core/io_priority_class.hh>
 #include <seastar/core/lowres_clock.hh>
@@ -50,9 +52,9 @@ class cluster_metadata_uploader_fixture
       , bucket(cloud_storage_clients::bucket_name("test-bucket")) {
         set_expectations_and_listen({});
         wait_for_controller_leadership().get();
-        tests::cooperative_spin_wait_with_timeout(5s, [this] {
+        RPTEST_REQUIRE_EVENTUALLY(5s, [this] {
             return app.storage.local().get_cluster_uuid().has_value();
-        }).get();
+        });
         cluster_uuid = app.storage.local().get_cluster_uuid().value();
     }
 
@@ -66,9 +68,23 @@ class cluster_metadata_uploader_fixture
         auto m_res = co_await download_highest_manifest_for_cluster(
           remote, cluster_uuid, bucket, retry_node);
         if (!m_res.has_value()) {
+            vlog(
+              logger.debug,
+              "Current manifest has id {}, waiting for > {}",
+              -1,
+              initial_meta_id);
             co_return false;
         }
         if (m_res.value().metadata_id <= initial_meta_id) {
+            vlog(
+              logger.debug,
+              "Current manifest has id {}, waiting for > {}",
+              m_res.value(),
+              initial_meta_id);
+            co_return false;
+        }
+        if (m_res.value().controller_snapshot_path.empty()) {
+            vlog(logger.debug, "Missing controller snapshot");
             co_return false;
         }
         *downloaded_manifest = std::move(m_res.value());
@@ -106,6 +122,7 @@ class cluster_metadata_uploader_fixture
     }
 
 protected:
+    scoped_config test_local_cfg;
     cluster::consensus_ptr raft0;
     cluster::controller_stm& controller_stm;
     cloud_storage::remote& remote;
@@ -115,8 +132,7 @@ class cluster_metadata_uploader_fixture
 
 FIXTURE_TEST(
   test_download_highest_manifest, cluster_metadata_uploader_fixture) {
-    cluster::cloud_metadata::uploader uploader(
-      cluster_uuid, bucket, remote, raft0);
+    auto& uploader = app.controller->metadata_uploader();
     retry_chain_node retry_node(
       never_abort, ss::lowres_clock::time_point::max(), 10ms);
 
@@ -155,8 +171,7 @@ FIXTURE_TEST(
 
 FIXTURE_TEST(
   test_download_highest_manifest_errors, cluster_metadata_uploader_fixture) {
-    cluster::cloud_metadata::uploader uploader(
-      cluster_uuid, bucket, remote, raft0);
+    auto& uploader = app.controller->metadata_uploader();
     retry_chain_node retry_node(
       never_abort, ss::lowres_clock::time_point::min(), 10ms);
     auto down_res
@@ -166,8 +181,7 @@ FIXTURE_TEST(
 }
 
 FIXTURE_TEST(test_upload_next_metadata, cluster_metadata_uploader_fixture) {
-    cluster::cloud_metadata::uploader uploader(
-      cluster_uuid, bucket, remote, raft0);
+    auto& uploader = app.controller->metadata_uploader();
     retry_chain_node retry_node(
       never_abort, ss::lowres_clock::time_point::max(), 10ms);
     RPTEST_REQUIRE_EVENTUALLY(5s, [this] { return raft0->is_leader(); });
@@ -264,10 +278,9 @@ FIXTURE_TEST(test_upload_in_term, cluster_metadata_uploader_fixture) {
     };
     const auto snap_offset = get_local_snap_offset();
 
-    config::shard_local_cfg()
-      .cloud_storage_cluster_metadata_upload_interval_ms.set_value(1000ms);
-    cluster::cloud_metadata::uploader uploader(
-      cluster_uuid, bucket, remote, raft0);
+    test_local_cfg.get("cloud_storage_cluster_metadata_upload_interval_ms")
+      .set_value(1000ms);
+    auto& uploader = app.controller->metadata_uploader();
     cluster::cloud_metadata::cluster_metadata_id highest_meta_id{0};
 
     // Checks that metadata is uploaded a new term, stepping down in between
@@ -280,13 +293,6 @@ FIXTURE_TEST(test_upload_in_term, cluster_metadata_uploader_fixture) {
 
           // Start uploading in this term.
           auto upload_in_term = uploader.upload_until_term_change();
-          auto defer = ss::defer([&] {
-              uploader.stop_and_wait().get();
-              try {
-                  upload_in_term.get();
-              } catch (...) {
-              }
-          });
 
           // Keep checking the latest manifest for whether the metadata ID is
           // some non-zero value (indicating we've uploaded multiple manifests);
@@ -305,7 +311,6 @@ FIXTURE_TEST(test_upload_in_term, cluster_metadata_uploader_fixture) {
           // Stop the upload loop and continue in a new term.
           raft0->step_down("forced stepdown").get();
           upload_in_term.get();
-          defer.cancel();
       };
     for (int i = 0; i < 3; ++i) {
         check_uploads_in_term_and_stepdown(snap_offset);
@@ -333,20 +338,12 @@ FIXTURE_TEST(
     // Write a snapshot and begin the upload loop.
     RPTEST_REQUIRE_EVENTUALLY(
       5s, [this] { return controller_stm.maybe_write_snapshot(); });
-    config::shard_local_cfg()
-      .cloud_storage_cluster_metadata_upload_interval_ms.set_value(1000ms);
-    cluster::cloud_metadata::uploader uploader(
-      cluster_uuid, bucket, remote, raft0);
+    test_local_cfg.get("cloud_storage_cluster_metadata_upload_interval_ms")
+      .set_value(1000ms);
+    auto& uploader = app.controller->metadata_uploader();
     RPTEST_REQUIRE_EVENTUALLY(5s, [this] { return raft0->is_leader(); });
 
     auto upload_in_term = uploader.upload_until_term_change();
-    auto defer = ss::defer([&] {
-        uploader.stop_and_wait().get();
-        try {
-            upload_in_term.get();
-        } catch (...) {
-        }
-    });
     // Wait for some valid metadata to show up.
     cluster::cloud_metadata::cluster_metadata_manifest manifest;
     RPTEST_REQUIRE_EVENTUALLY(5s, [this, &manifest] {
@@ -380,3 +377,27 @@ FIXTURE_TEST(
         return num_deletes >= 2;
     });
 }
+
+FIXTURE_TEST(test_run_loop, cluster_metadata_uploader_fixture) {
+    RPTEST_REQUIRE_EVENTUALLY(
+      5s, [this] { return controller_stm.maybe_write_snapshot(); });
+    test_local_cfg.get("cloud_storage_cluster_metadata_upload_interval_ms")
+      .set_value(1000ms);
+    auto& uploader = app.controller->metadata_uploader();
+    // Run the upload loop and make sure that new leaders continue to upload.
+    uploader.start();
+    cluster::cloud_metadata::cluster_metadata_id highest_meta_id{-1};
+    for (int i = 0; i < 3; i++) {
+        auto initial_meta_id = highest_meta_id;
+        cluster::cloud_metadata::cluster_metadata_manifest manifest;
+        RPTEST_REQUIRE_EVENTUALLY(10s, [&]() -> ss::future<bool> {
+            return downloaded_manifest_has_higher_id(
+              initial_meta_id, &manifest);
+        });
+        BOOST_REQUIRE_GT(manifest.metadata_id, highest_meta_id);
+        highest_meta_id = manifest.metadata_id;
+
+        // Stop the upload loop and continue in a new term.
+        raft0->step_down("forced stepdown").get();
+    }
+}