cl: Hook in remote_cluster_connection to link

Signed-off-by: Michael Boquard <[email protected]>

Author: michael-redpanda

src/v/cluster_link/deps.cc

@@ -145,69 +145,8 @@ class remote_cluster_connection_factory_impl
               link_id);
             return nullptr;
         }
-        const auto& link_connection = metadata->get().connection;
-        kafka::client::connection_configuration config;
-        config.initial_brokers = link_connection.bootstrap_servers;
-        if (
-          link_connection.ca.has_value() || link_connection.cert.has_value()
-          || link_connection.key.has_value()) {
-            kafka::client::tls_configuration tls_cfg;
-            if (link_connection.ca.has_value()) {
-                ss::visit(
-                  link_connection.ca.value(),
-                  [&](const model::tls_file_path& path) {
-                      tls_cfg.truststore = std::filesystem::path(path());
-                  },
-                  [&](const model::tls_value& value) {
-                      tls_cfg.truststore = value();
-                  });
-            }
-            if (
-              link_connection.cert.has_value()
-              && link_connection.key.has_value()) {
-                kafka::client::key_store k_store;
-                ss::visit(
-                  link_connection.cert.value(),
-                  [&](const model::tls_file_path& path) {
-                      vassert(
-                        std::holds_alternative<model::tls_file_path>(
-                          link_connection.key.value()),
-                        "Key must be a file path if cert is a file path");
-                      k_store = kafka::client::key_cert_path{
-                        .key = std::filesystem::path(path()),
-                        .cert = std::filesystem::path(
-                          std::get<model::tls_file_path>(
-                            link_connection.key.value())())};
-                  },
-                  [&](const model::tls_value& value) {
-                      vassert(
-                        std::holds_alternative<model::tls_value>(
-                          link_connection.key.value()),
-                        "Key must be a value if cert is a value");
-                      k_store = kafka::client::key_cert{
-                        .key = value(),
-                        .cert = std::get<model::tls_value>(
-                          link_connection.key.value())()};
-                  });
-                tls_cfg.k_store = std::move(k_store);
-            }
-            config.broker_tls = std::move(tls_cfg);
-        }
-        if (link_connection.authn_config.has_value()) {
-            ss::visit(
-              link_connection.authn_config.value(),
-              [&](const model::scram_credentials& creds) {
-                  config.sasl_cfg = kafka::client::sasl_configuration{
-                    .mechanism = creds.mechanism,
-                    .username = creds.username,
-                    .password = creds.password};
-              });
-        }
-
-        config.client_id = ssx::sformat(
-          "cluster-link-{}-{}", metadata->get().name, metadata->get().uuid);
-
-        return remote_cluster_connection::make_default(std::move(config));
+        return remote_cluster_connection::make_default(
+          make_configuration(metadata->get()));
     }
 
 private:
@@ -232,4 +171,71 @@ remote_cluster_connection_factory::make_default(
     return std::make_unique<remote_cluster_connection_factory_impl>(
       std::move(link_registry));
 }
+
+kafka::client::connection_configuration
+remote_cluster_connection_factory::make_configuration(
+  const model::metadata& md) {
+    const auto& link_connection = md.connection;
+
+    kafka::client::connection_configuration config;
+    config.initial_brokers = link_connection.bootstrap_servers;
+    if (
+      link_connection.ca.has_value() || link_connection.cert.has_value()
+      || link_connection.key.has_value()) {
+        kafka::client::tls_configuration tls_cfg;
+        if (link_connection.ca.has_value()) {
+            ss::visit(
+              link_connection.ca.value(),
+              [&](const model::tls_file_path& path) {
+                  tls_cfg.truststore = std::filesystem::path(path());
+              },
+              [&](const model::tls_value& value) {
+                  tls_cfg.truststore = value();
+              });
+        }
+        if (
+          link_connection.cert.has_value() && link_connection.key.has_value()) {
+            kafka::client::key_store k_store;
+            ss::visit(
+              link_connection.cert.value(),
+              [&](const model::tls_file_path& path) {
+                  vassert(
+                    std::holds_alternative<model::tls_file_path>(
+                      link_connection.key.value()),
+                    "Key must be a file path if cert is a file path");
+                  k_store = kafka::client::key_cert_path{
+                    .key = std::filesystem::path(path()),
+                    .cert = std::filesystem::path(
+                      std::get<model::tls_file_path>(
+                        link_connection.key.value())())};
+              },
+              [&](const model::tls_value& value) {
+                  vassert(
+                    std::holds_alternative<model::tls_value>(
+                      link_connection.key.value()),
+                    "Key must be a value if cert is a value");
+                  k_store = kafka::client::key_cert{
+                    .key = value(),
+                    .cert = std::get<model::tls_value>(
+                      link_connection.key.value())()};
+              });
+            tls_cfg.k_store = std::move(k_store);
+        }
+        config.broker_tls = std::move(tls_cfg);
+    }
+    if (link_connection.authn_config.has_value()) {
+        ss::visit(
+          link_connection.authn_config.value(),
+          [&](const model::scram_credentials& creds) {
+              config.sasl_cfg = kafka::client::sasl_configuration{
+                .mechanism = creds.mechanism,
+                .username = creds.username,
+                .password = creds.password};
+          });
+    }
+
+    config.client_id = ssx::sformat("cluster-link-{}-{}", md.name, md.uuid);
+
+    return config;
+}
 } // namespace cluster_link

src/v/cluster_link/deps.h

@@ -100,6 +100,9 @@ class remote_cluster_connection_factory {
 
     static std::unique_ptr<remote_cluster_connection_factory>
     make_default(std::unique_ptr<link_registry> link_registry);
+
+    static kafka::client::connection_configuration
+    make_configuration(const model::metadata&);
 };
 
 /**
@@ -117,6 +120,7 @@ class link_factory {
 
     virtual std::unique_ptr<link> create_link(
       ::model::node_id self,
+      model::id_t link_id,
       model::metadata config,
       kafka::data::rpc::partition_leader_cache* partition_leader_cache,
       kafka::data::rpc::partition_manager* partition_manager)

src/v/cluster_link/link.cc

@@ -52,11 +52,13 @@ link::link(
   ss::lowres_clock::duration task_reconciler_interval,
   model::metadata config,
   partition_leader_cache* partition_leader_cache,
-  partition_manager* partition_manager)
+  partition_manager* partition_manager,
+  std::unique_ptr<remote_cluster_connection> remote_cluster)
   : _self(self)
   , _config(std::move(config))
   , _partition_leader_cache(partition_leader_cache)
   , _partition_manager(partition_manager)
+  , _remote_cluster(std::move(remote_cluster))
   , _task_reconciler_interval(task_reconciler_interval) {}
 
 ss::future<> link::start() {

src/v/cluster_link/link.h

@@ -11,6 +11,7 @@
 
 #pragma once
 
+#include "cluster_link/deps.h"
 #include "cluster_link/model/types.h"
 #include "cluster_link/task.h"
 #include "cluster_link/types.h"
@@ -29,7 +30,8 @@ class link {
       ss::lowres_clock::duration task_reconciler_interval,
       model::metadata config,
       kafka::data::rpc::partition_leader_cache* partition_leader_cache,
-      kafka::data::rpc::partition_manager* partition_manager);
+      kafka::data::rpc::partition_manager* partition_manager,
+      std::unique_ptr<remote_cluster_connection> remote_cluster);
     link(const link&) = delete;
     link(link&&) = delete;
     link& operator=(const link&) = delete;
@@ -79,6 +81,7 @@ class link {
     model::metadata _config;
     kafka::data::rpc::partition_leader_cache* _partition_leader_cache;
     kafka::data::rpc::partition_manager* _partition_manager;
+    std::unique_ptr<remote_cluster_connection> _remote_cluster;
     notification_list<task_state_change_cb, task_state_notification_id>
       _task_state_change_notifications;
     ss::lowres_clock::duration _task_reconciler_interval;

src/v/cluster_link/manager.cc

@@ -130,6 +130,7 @@ ss::future<> manager::handle_on_link_change(model::id_t id) {
             auto units = co_await _link_task_reconciler_mutex.get_units(_as);
             auto new_link = _link_factory->create_link(
               _self,
+              id,
               link_metadata.copy(),
               _partition_leader_cache.get(),
               _partition_manager.get());

src/v/cluster_link/service.cc

@@ -13,6 +13,7 @@
 
 #include "cluster/cluster_link/frontend.h"
 #include "cluster/partition_manager.h"
+#include "cluster_link/deps.h"
 #include "cluster_link/link.h"
 #include "cluster_link/logger.h"
 #include "cluster_link/manager.h"
@@ -52,18 +53,35 @@ class link_registry_adapter : public link_registry {
 class default_link_factory : public link_factory {
 public:
     static constexpr auto link_reconciler_period = 5min;
+    explicit default_link_factory(
+      std::unique_ptr<remote_cluster_connection_factory> rccf)
+      : _rccf(std::move(rccf)) {}
+
     std::unique_ptr<link> create_link(
       ::model::node_id self,
+      model::id_t link_id,
       model::metadata config,
       partition_leader_cache* partition_leader_cache,
       partition_manager* partition_manager) override {
+        auto remote_cluster = _rccf->make_remote_cluster_connection(link_id);
+        if (!remote_cluster) {
+            vlog(
+              cllog.error,
+              "Failed to create remote cluster connection for link {}",
+              link_id);
+            return nullptr;
+        }
         return std::make_unique<link>(
           self,
           link_reconciler_period,
           std::move(config),
           partition_leader_cache,
-          partition_manager);
+          partition_manager,
+          std::move(remote_cluster));
     }
+
+private:
+    std::unique_ptr<remote_cluster_connection_factory> _rccf;
 };
 
 service::service(
@@ -92,7 +110,9 @@ ss::future<> service::start() {
       partition_manager::make_default(
         _shard_table, _partition_manager, _smp_group),
       std::make_unique<link_registry_adapter>(&_plf->local()),
-      std::make_unique<default_link_factory>(),
+      std::make_unique<default_link_factory>(
+        remote_cluster_connection_factory::make_default(
+          std::make_unique<link_registry_adapter>(&_plf->local()))),
       30s); // Temporary until we have a proper configuration for this
 
     // Register notifications before the manager starts.  The manager will have

src/v/cluster_link/tests/BUILD

@@ -30,7 +30,9 @@ redpanda_cc_gtest(
         ":test_deps",
         "//src/v/cluster",
         "//src/v/cluster/cluster_link/tests:test_lib",
+        "//src/v/cluster_link:errc",
         "//src/v/cluster_link:impl",
+        "//src/v/kafka/client:configuration",
         "//src/v/model",
         "//src/v/test_utils:gtest",
         "@abseil-cpp//absl/container:flat_hash_map",

src/v/cluster_link/tests/deps.h

@@ -12,12 +12,67 @@
 #pragma once
 
 #include "cluster/cluster_link/table.h"
+#include "cluster_link/deps.h"
 #include "cluster_link/manager.h"
 #include "kafka/data/rpc/deps.h"
 
 #include <seastar/util/defer.hh>
 namespace cluster_link::tests {
 
+class unimplemented_connection : public remote_cluster_connection {
+public:
+    explicit unimplemented_connection(
+      kafka::client::connection_configuration config)
+      : _config(std::move(config)) {}
+
+    ss::future<err_info> start() override { co_return err_info(errc::success); }
+    ss::future<> stop() override { return ss::now(); }
+
+    kafka::client::topic_cache& get_topics() override {
+        throw std::runtime_error("not implemented");
+    }
+
+    ss::future<> request_metadata_update() override {
+        throw std::runtime_error("not implemented");
+    }
+
+    bool is_connected() const override { return true; }
+
+    ss::future<kafka::describe_configs_response> describe_topics(
+      chunked_vector<::model::topic>, chunked_vector<ss::sstring>) override {
+        throw std::runtime_error("not implemented");
+    }
+
+    const kafka::client::connection_configuration& get_config() const {
+        return _config;
+    }
+
+private:
+    kafka::client::connection_configuration _config;
+};
+
+class unimplemented_connection_factory
+  : public remote_cluster_connection_factory {
+public:
+    explicit unimplemented_connection_factory(std::unique_ptr<link_registry> lr)
+      : _link_registry(std::move(lr)) {}
+
+    std::unique_ptr<remote_cluster_connection>
+    make_remote_cluster_connection(model::id_t link_id) override {
+        auto metadata = _link_registry->find_link_by_id(link_id);
+        if (!metadata) {
+            throw std::runtime_error(
+              fmt::format("No link found for id {}", link_id()));
+        }
+        return std::make_unique<unimplemented_connection>(
+          remote_cluster_connection_factory::make_configuration(
+            metadata->get()));
+    }
+
+private:
+    std::unique_ptr<link_registry> _link_registry;
+};
+
 class test_link_registry : public link_registry {
 public:
     explicit test_link_registry(cluster::cluster_link::table* table)
@@ -195,6 +250,8 @@ class cluster_link_manager_test_fixture {
 
     link_factory* get_link_factory() { return _lf; }
 
+    ss::sharded<cluster::cluster_link::table>& get_table() { return _table; }
+
 private:
     chunked_vector<ss::deferred_action<ss::noncopyable_function<void()>>>
       _notification_cleanups;
@@ -208,4 +265,25 @@ class cluster_link_manager_test_fixture {
     ::model::node_id _self;
     model::id_t _next_link_id{0};
 };
+
+class test_fixture_rccf : public remote_cluster_connection_factory {
+public:
+    explicit test_fixture_rccf(cluster_link_manager_test_fixture* clmtf)
+      : _clmtf(clmtf) {}
+
+    std::unique_ptr<remote_cluster_connection>
+    make_remote_cluster_connection(model::id_t link_id) override {
+        auto md = _clmtf->get_table().local().find_link_by_id(link_id);
+        if (!md) {
+            throw std::runtime_error(
+              fmt::format("No link found for id {}", link_id()));
+        }
+
+        return std::make_unique<unimplemented_connection>(
+          remote_cluster_connection_factory::make_configuration(md->get()));
+    }
+
+private:
+    cluster_link_manager_test_fixture* _clmtf;
+};
 } // namespace cluster_link::tests