cloud_storage: remote labels

src/v/archival/archival_metadata_stm.cc

@@ -15,6 +15,7 @@
 #include "bytes/iostream.h"
 #include "cloud_storage/partition_manifest.h"
 #include "cloud_storage/remote.h"
+#include "cloud_storage/remote_path_provider.h"
 #include "cloud_storage/types.h"
 #include "cluster/errc.h"
 #include "cluster/logger.h"
@@ -641,14 +642,16 @@ archival_metadata_stm::archival_metadata_stm(
   raft::consensus* raft,
   cloud_storage::remote& remote,
   features::feature_table& ft,
-  ss::logger& logger)
+  ss::logger& logger,
+  std::optional<cloud_storage::remote_label> remote_label)
   : raft::persisted_stm<>(archival_stm_snapshot, logger, raft)
   , _logger(logger, ssx::sformat("ntp: {}", raft->ntp()))
   , _mem_tracker(ss::make_shared<util::mem_tracker>(raft->ntp().path()))
   , _manifest(ss::make_shared<cloud_storage::partition_manifest>(
       raft->ntp(), raft->log_config().get_initial_revision(), _mem_tracker))
   , _cloud_storage_api(remote)
-  , _feature_table(ft) {}
+  , _feature_table(ft)
+  , _remote_path_provider({remote_label}) {}
 
 ss::future<std::error_code> archival_metadata_stm::truncate(
   model::offset start_rp_offset,
@@ -1687,10 +1690,12 @@ archival_metadata_stm::state_dirty archival_metadata_stm::get_dirty(
 archival_metadata_stm_factory::archival_metadata_stm_factory(
   bool cloud_storage_enabled,
   ss::sharded<cloud_storage::remote>& cloud_storage_api,
-  ss::sharded<features::feature_table>& feature_table)
+  ss::sharded<features::feature_table>& feature_table,
+  ss::sharded<cluster::topic_table>& topics)
   : _cloud_storage_enabled(cloud_storage_enabled)
   , _cloud_storage_api(cloud_storage_api)
-  , _feature_table(feature_table) {}
+  , _feature_table(feature_table)
+  , _topics(topics) {}
 
 bool archival_metadata_stm_factory::is_applicable_for(
   const storage::ntp_config& ntp_cfg) const {
@@ -1700,8 +1705,18 @@ bool archival_metadata_stm_factory::is_applicable_for(
 
 void archival_metadata_stm_factory::create(
   raft::state_machine_manager_builder& builder, raft::consensus* raft) {
+    auto topic_md = _topics.local().get_topic_metadata_ref(
+      model::topic_namespace_view(raft->ntp()));
+    auto remote_label
+      = topic_md.has_value()
+          ? topic_md->get().get_configuration().properties.remote_label
+          : std::nullopt;
     auto stm = builder.create_stm<cluster::archival_metadata_stm>(
-      raft, _cloud_storage_api.local(), _feature_table.local(), clusterlog);
+      raft,
+      _cloud_storage_api.local(),
+      _feature_table.local(),
+      clusterlog,
+      remote_label);
     raft->log()->stm_manager()->add_stm(stm);
 }
 

src/v/archival/archival_metadata_stm.h

@@ -13,9 +13,11 @@
 
 #include "cloud_storage/fwd.h"
 #include "cloud_storage/partition_manifest.h"
+#include "cloud_storage/remote_path_provider.h"
 #include "cloud_storage/types.h"
 #include "cluster/errc.h"
 #include "cluster/state_machine_registry.h"
+#include "cluster/topic_table.h"
 #include "features/fwd.h"
 #include "model/fundamental.h"
 #include "model/record.h"
@@ -133,7 +135,8 @@ class archival_metadata_stm final : public raft::persisted_stm<> {
       raft::consensus*,
       cloud_storage::remote& remote,
       features::feature_table&,
-      ss::logger& logger);
+      ss::logger& logger,
+      std::optional<cloud_storage::remote_label>);
 
     /// Add segments to the raft log, replicate them and
     /// wait until it is applied to the STM.
@@ -277,6 +280,10 @@ class archival_metadata_stm final : public raft::persisted_stm<> {
         return _compacted_replaced_bytes;
     }
 
+    const cloud_storage::remote_path_provider& path_provider() const {
+        return _remote_path_provider;
+    }
+
 private:
     ss::future<bool>
     do_sync(model::timeout_clock::duration timeout, ss::abort_source* as);
@@ -374,6 +381,7 @@ class archival_metadata_stm final : public raft::persisted_stm<> {
 
     cloud_storage::remote& _cloud_storage_api;
     features::feature_table& _feature_table;
+    const cloud_storage::remote_path_provider _remote_path_provider;
     ss::abort_source _download_as;
 
     // for observability: keep track of the number of cloud bytes "removed" by
@@ -389,7 +397,8 @@ class archival_metadata_stm_factory : public state_machine_factory {
     archival_metadata_stm_factory(
       bool cloud_storage_enabled,
       ss::sharded<cloud_storage::remote>&,
-      ss::sharded<features::feature_table>&);
+      ss::sharded<features::feature_table>&,
+      ss::sharded<cluster::topic_table>&);
 
     bool is_applicable_for(const storage::ntp_config&) const final;
     void create(raft::state_machine_manager_builder&, raft::consensus*) final;
@@ -398,6 +407,7 @@ class archival_metadata_stm_factory : public state_machine_factory {
     bool _cloud_storage_enabled;
     ss::sharded<cloud_storage::remote>& _cloud_storage_api;
     ss::sharded<features::feature_table>& _feature_table;
+    ss::sharded<topic_table>& _topics;
 };
 
 } // namespace cluster

src/v/archival/tests/archival_metadata_stm_gtest.cc

@@ -102,7 +102,8 @@ class archival_metadata_stm_gtest_fixture : public raft::raft_fixture {
               node->raft().get(),
               stm_node.remote.local(),
               node->get_feature_table().local(),
-              fixture_logger);
+              fixture_logger,
+              std::nullopt);
 
             stm_node.archival_stm = std::move(stm);
 

src/v/archival/tests/archival_metadata_stm_test.cc

@@ -140,7 +140,11 @@ struct archival_metadata_stm_fixture : archival_metadata_stm_base_fixture {
         create_raft();
         raft::state_machine_manager_builder builder;
         archival_stm = builder.create_stm<cluster::archival_metadata_stm>(
-          _raft.get(), cloud_api.local(), feature_table.local(), logger);
+          _raft.get(),
+          cloud_api.local(),
+          feature_table.local(),
+          logger,
+          std::nullopt);
 
         _raft->start(std::move(builder)).get();
         _started = true;
@@ -351,7 +355,11 @@ FIXTURE_TEST(test_snapshot_loading, archival_metadata_stm_base_fixture) {
 
     raft::state_machine_manager_builder builder;
     auto archival_stm = builder.create_stm<cluster::archival_metadata_stm>(
-      _raft.get(), cloud_api.local(), feature_table.local(), logger);
+      _raft.get(),
+      cloud_api.local(),
+      feature_table.local(),
+      logger,
+      std::nullopt);
     _raft->start(std::move(builder)).get();
     _started = true;
     wait_for_confirmed_leader();
@@ -446,7 +454,11 @@ FIXTURE_TEST(test_sname_derivation, archival_metadata_stm_base_fixture) {
 
     raft::state_machine_manager_builder builder;
     auto archival_stm = builder.create_stm<cluster::archival_metadata_stm>(
-      _raft.get(), cloud_api.local(), feature_table.local(), logger);
+      _raft.get(),
+      cloud_api.local(),
+      feature_table.local(),
+      logger,
+      std::nullopt);
 
     _raft->start(std::move(builder)).get();
     _started = true;
@@ -655,7 +667,11 @@ FIXTURE_TEST(
 
     raft::state_machine_manager_builder builder;
     auto archival_stm = builder.create_stm<cluster::archival_metadata_stm>(
-      _raft.get(), cloud_api.local(), feature_table.local(), logger);
+      _raft.get(),
+      cloud_api.local(),
+      feature_table.local(),
+      logger,
+      std::nullopt);
     _raft->start(std::move(builder)).get();
     _started = true;
     wait_for_confirmed_leader();

src/v/redpanda/application.cc

@@ -2685,7 +2685,8 @@ void application::start_runtime_services(
           pm.register_factory<cluster::archival_metadata_stm_factory>(
             config::shard_local_cfg().cloud_storage_enabled(),
             cloud_storage_api,
-            feature_table);
+            feature_table,
+            controller->get_topics_state());
           pm.register_factory<kafka::group_tx_tracker_stm_factory>();
       })
       .get();