shadowing: add link report cache

src/v/cluster_link/BUILD

@@ -185,6 +185,30 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "shadow_link_report_cache",
+    srcs = [
+        "shadow_link_report_cache.cc",
+    ],
+    hdrs = [
+        "shadow_link_report_cache.h",
+    ],
+    implementation_deps = [
+        ":logger",
+        "//src/v/ssx:future_util",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":cluster_link",
+        "//src/v/cluster_link/model",
+        "//src/v/config",
+        "//src/v/model",
+        "//src/v/ssx:semaphore",
+        "@abseil-cpp//absl/container:flat_hash_map",
+        "@seastar",
+    ],
+)
+
 redpanda_cc_library(
     name = "cluster_link",
     srcs = [

src/v/cluster_link/errc.cc

@@ -82,6 +82,10 @@ struct error_category final : public std::error_category {
             return "failed to stop task";
         case errc::failed_to_pause_task:
             return "failed to pause task";
+        case errc::report_generation_timed_out:
+            return "report generation timed out";
+        case errc::report_generation_unknown_error:
+            return "report generation unknown error";
         }
 
         return "(unknown error code)";

src/v/cluster_link/errc.h

@@ -49,6 +49,8 @@ enum class errc : int {
     link_verification_unknown_error,
     failed_to_stop_task,
     failed_to_pause_task,
+    report_generation_timed_out,
+    report_generation_unknown_error,
 };
 
 std::error_code make_error_code(errc) noexcept;

src/v/cluster_link/fwd.h

@@ -15,4 +15,6 @@ namespace cluster_link {
 class link;
 class manager;
 class service;
+class shadow_link_report_cache;
+class shadow_link_report_fetcher_impl;
 } // namespace cluster_link

src/v/cluster_link/model/BUILD

@@ -29,6 +29,7 @@ redpanda_cc_library(
         "//src/v/serde:optional",
         "//src/v/serde:variant",
         "//src/v/serde:vector",
+        "//src/v/ssx:async_algorithm",
         "//src/v/utils:absl_sstring_hash",
         "//src/v/utils:named_type",
         "//src/v/utils:unresolved_address",

src/v/cluster_link/model/types.cc

@@ -13,6 +13,7 @@
 
 #include "base/format_to.h"
 #include "model/timestamp.h"
+#include "ssx/async_algorithm.h"
 #include "utils/to_string.h"
 
 #include <seastar/util/variant_utils.hh>
@@ -337,6 +338,20 @@ shadow_link_status_report_request::format_to(fmt::iterator it) const {
     return fmt::format_to(it, "{{ link_id: {} }}", link_id);
 }
 
+ss::future<shadow_link_status_topic_response>
+shadow_link_status_topic_response::copy() const {
+    shadow_link_status_topic_response copy;
+    copy.status = status;
+    copy.partition_reports.reserve(partition_reports.size());
+    ssx::async_counter counter;
+    co_await ssx::async_for_each_counter(
+      counter, partition_reports, [&copy](const auto& pid_report) {
+          const auto& [pid, report] = pid_report;
+          copy.partition_reports.emplace(pid, report);
+      });
+    co_return copy;
+}
+
 fmt::iterator
 shadow_link_status_topic_response::format_to(fmt::iterator it) const {
     return fmt::format_to(
@@ -376,6 +391,23 @@ shadow_link_status_report_response::format_to(fmt::iterator it) const {
 } // namespace cluster_link::rpc
 
 namespace cluster_link::model {
+
+ss::future<shadow_link_status_report> shadow_link_status_report::copy() const {
+    shadow_link_status_report copy;
+    copy.link_id = link_id;
+
+    copy.topic_responses.reserve(topic_responses.size());
+    for (const auto& [topic, response] : topic_responses) {
+        copy.topic_responses.emplace(topic, co_await response.copy());
+    }
+
+    copy.task_status_reports.reserve(task_status_reports.size());
+    for (const auto& [task_name, reports] : task_status_reports) {
+        copy.task_status_reports.emplace(task_name, reports.copy());
+    }
+    co_return copy;
+}
+
 fmt::iterator shadow_link_status_report::format_to(fmt::iterator it) const {
     return fmt::format_to(
       it,

src/v/cluster_link/model/types.h

@@ -1193,6 +1193,8 @@ struct shadow_link_status_topic_response
 
     fmt::iterator format_to(fmt::iterator) const;
 
+    ss::future<shadow_link_status_topic_response> copy() const;
+
     auto serde_fields() { return std::tie(status, partition_reports); }
 };
 
@@ -1234,6 +1236,8 @@ struct shadow_link_status_report {
     chunked_hash_map<ss::sstring, chunked_vector<model::task_status_report>>
       task_status_reports;
 
+    ss::future<shadow_link_status_report> copy() const;
+
     fmt::iterator format_to(fmt::iterator) const;
 };
 

src/v/cluster_link/service.h

@@ -171,6 +171,8 @@ class service : public ss::peering_sharded_service<service> {
     ss::future<model::status_report_ret_t>
     shadow_link_report(model::name_t name);
 
+    const manager& get_manager() const { return *_manager; }
+
 private:
     void register_notifications();
     void unregister_notifications();