Merge pull request redpanda-data#21582 from WillemKauf/guardrail_mani…

…fest

cloud_storage: add `topic_mount_manifest`

src/v/cloud_storage/BUILD

@@ -56,6 +56,8 @@ redpanda_cc_library(
         "segment_state.cc",
         "topic_manifest.cc",
         "topic_manifest_downloader.cc",
+        "topic_mount_handler.cc",
+        "topic_mount_manifest.cc",
         "topic_path_utils.cc",
         "tx_range_manifest.cc",
         "types.cc",
@@ -105,6 +107,8 @@ redpanda_cc_library(
         "spillover_manifest.h",
         "topic_manifest.h",
         "topic_manifest_downloader.h",
+        "topic_mount_handler.h",
+        "topic_mount_manifest.h",
         "topic_path_utils.h",
         "tx_range_manifest.h",
         "types.h",

src/v/cloud_storage/CMakeLists.txt

@@ -17,6 +17,7 @@ v_cc_library(
     access_time_tracker.cc
     cache_probe.cc
     download_exception.cc
+    topic_mount_manifest.cc
     partition_manifest.cc
     partition_manifest_downloader.cc
     partition_path_utils.cc
@@ -44,6 +45,7 @@ v_cc_library(
     segment_path_utils.cc
     topic_manifest.cc
     topic_manifest_downloader.cc
+    topic_mount_handler.cc
     topic_path_utils.cc
     async_manifest_view.cc
     materialized_manifest_cache.cc

src/v/cloud_storage/base_manifest.cc

@@ -31,6 +31,9 @@ std::ostream& operator<<(std::ostream& s, manifest_type t) {
     case manifest_type::spillover:
         s << "spillover";
         break;
+    case manifest_type::topic_mount:
+        s << "topic_mount";
+        break;
     }
     return s;
 }

src/v/cloud_storage/base_manifest.h

@@ -29,6 +29,7 @@ enum class manifest_type {
     tx_range,
     cluster_metadata,
     spillover,
+    topic_mount
 };
 
 std::ostream& operator<<(std::ostream& s, manifest_type t);

src/v/cloud_storage/fwd.h

@@ -19,6 +19,7 @@ class remote_partition;
 class remote_path_provider;
 class remote_segment;
 class partition_manifest;
+class topic_mount_manifest;
 class topic_manifest;
 class partition_probe;
 class async_manifest_view;

src/v/cloud_storage/remote.cc

@@ -277,6 +277,9 @@ ss::future<download_result> remote::do_download_manifest(
                 case manifest_type::spillover:
                     _probe.spillover_manifest_download();
                     break;
+                case manifest_type::topic_mount:
+                    _probe.topic_mount_manifest_download();
+                    break;
                 }
                 co_return download_result::success;
             } catch (...) {
@@ -383,6 +386,9 @@ ss::future<upload_result> remote::upload_manifest(
             case manifest_type::spillover:
                 _probe.spillover_manifest_upload();
                 break;
+            case manifest_type::topic_mount:
+                _probe.topic_mount_manifest_upload();
+                break;
             }
             _probe.register_upload_size(size);
             co_return upload_result::success;

src/v/cloud_storage/remote_path_provider.cc

@@ -13,6 +13,7 @@
 #include "cloud_storage/remote_label.h"
 #include "cloud_storage/segment_path_utils.h"
 #include "cloud_storage/spillover_manifest.h"
+#include "cloud_storage/topic_mount_manifest.h"
 #include "cloud_storage/topic_path_utils.h"
 #include "cloud_storage/types.h"
 #include "model/fundamental.h"
@@ -116,6 +117,14 @@ ss::sstring remote_path_provider::spillover_manifest_path(
       spillover_manifest::filename(c));
 }
 
+ss::sstring remote_path_provider::topic_mount_manifest_path(
+  const topic_mount_manifest& manifest) const {
+    return fmt::format(
+      "migration/{}/{}",
+      manifest.get_source_label().cluster_uuid,
+      manifest.get_tp_ns().path());
+}
+
 ss::sstring remote_path_provider::segment_path(
   const model::ntp& ntp,
   model::initial_revision_id rev,

src/v/cloud_storage/remote_path_provider.h

@@ -82,6 +82,9 @@ class remote_path_provider {
       const partition_manifest& stm_manifest,
       const spillover_manifest_path_components& c) const;
 
+    ss::sstring
+    topic_mount_manifest_path(const topic_mount_manifest& manifest) const;
+
     // Segment paths.
     ss::sstring segment_path(
       const partition_manifest& manifest, const segment_meta& segment) const;

src/v/cloud_storage/remote_probe.h

@@ -117,6 +117,24 @@ class remote_probe {
         return _cnt_tx_manifest_downloads;
     }
 
+    /// Register manifest (re)upload
+    void topic_mount_manifest_upload() { _cnt_topic_mount_manifest_uploads++; }
+
+    /// Get manifest (re)upload
+    uint64_t get_topic_mount_manifest_uploads() const {
+        return _cnt_topic_mount_manifest_uploads;
+    }
+
+    /// Register manifest download
+    void topic_mount_manifest_download() {
+        _cnt_topic_mount_manifest_downloads++;
+    }
+
+    /// Get manifest download
+    uint64_t get_topic_mount_manifest_downloads() const {
+        return _cnt_topic_mount_manifest_downloads;
+    }
+
     /// Register backof invocation during manifest upload
     void manifest_upload_backoff() { _cnt_manifest_upload_backoff++; }
 
@@ -295,6 +313,10 @@ class remote_probe {
     uint64_t _cnt_spillover_manifest_uploads{0};
     /// Number of spillover manifest downloads
     uint64_t _cnt_spillover_manifest_downloads{0};
+    /// Number of topic_mount manifest uploads
+    uint64_t _cnt_topic_mount_manifest_uploads{0};
+    /// Number of topic_mount manifest downloads
+    uint64_t _cnt_topic_mount_manifest_downloads{0};
 
     hist_t _client_acquisition_latency;
     hist_t _segment_download_latency;

src/v/cloud_storage/tests/CMakeLists.txt

@@ -33,6 +33,25 @@ rp_test(
   LABELS cloud_storage
 )
 
+rp_test(
+  UNIT_TEST
+  GTEST
+  BINARY_NAME gtest_cloud_storage
+  SOURCES
+    topic_mount_manifest_test.cc
+  LIBRARIES
+    v::gtest_main
+    v::seastar_testing_main
+    v::cloud_storage
+    v::storage_test_utils
+    v::cloud_roles
+    v::application
+    v::kafka_test_utils
+    v::http_test_utils
+  ARGS "-- -c 1"
+  LABELS cloud_storage
+)
+
 rp_test(
   UNIT_TEST
   BINARY_NAME topic_manifest
@@ -85,6 +104,7 @@ rp_test(
     s3_imposter.cc
     topic_manifest_downloader_test.cc
     topic_namespace_override_recovery_test.cc
+    topic_mount_handler_test.cc
     util.cc
   LIBRARIES
     v::gtest_main

src/v/cloud_storage/tests/remote_path_provider_test.cc

@@ -10,6 +10,7 @@
 #include "cloud_storage/partition_manifest.h"
 #include "cloud_storage/remote_path_provider.h"
 #include "cloud_storage/spillover_manifest.h"
+#include "cloud_storage/topic_mount_manifest.h"
 #include "cloud_storage/topic_path_utils.h"
 #include "cloud_storage/types.h"
 #include "gtest/gtest.h"
@@ -379,6 +380,14 @@ TEST_P(
                    .starts_with(partition_prefix));
 }
 
+TEST(RemotePathProviderTest, TestTopicMountManifestPath) {
+    remote_path_provider path_provider(test_label, std::nullopt);
+    topic_mount_manifest manifest(test_label, test_tp_ns);
+    EXPECT_STREQ(
+      path_provider.topic_mount_manifest_path(manifest).c_str(),
+      "migration/deadbeef-0000-0000-0000-000000000000/kafka/tp");
+}
+
 INSTANTIATE_TEST_SUITE_P(
   LabeledWithOverride,
   OverrideParamRemotePathProviderWithLabelTest,