cluster_link: export Schema Registry sync totals as prometheus metrics

src/v/cluster_link/schema_registry_sync/BUILD

@@ -57,11 +57,13 @@ redpanda_cc_library(
     name = "mirroring_task",
     srcs = [
         "mirroring_task.cc",
+        "probe.cc",
         "reconciler.cc",
         "scope.cc",
     ],
     hdrs = [
         "mirroring_task.h",
+        "probe.h",
         "reconciler.h",
         "scope.h",
     ],
@@ -84,6 +86,7 @@ redpanda_cc_library(
         ":source_reader",
         "//src/v/cluster_link:impl",
         "//src/v/cluster_link/model",
+        "//src/v/metrics",
         "//src/v/model",
         "//src/v/schema:registry",
         "//src/v/ssx:semaphore",

src/v/cluster_link/schema_registry_sync/mirroring_task.cc

@@ -139,8 +139,23 @@ void mirroring_task::reset_sync_state() {
     _last_full_sync.reset();
 }
 
+ss::future<cl_result<void>> mirroring_task::start() {
+    auto res = co_await task::start();
+    // The series stay registered for the task's whole non-stopped life,
+    // including while paused; stop() removes them. See probe.h for the
+    // lifecycle rationale. setup is idempotent, so resuming from paused
+    // (which also lands here) is fine.
+    if (res.has_value()) {
+        _probe.setup(get_link()->get_config()->name, [this] {
+            return get_live_sync_status().totals_since_task_start;
+        });
+    }
+    co_return res;
+}
+
 ss::future<cl_result<void>> mirroring_task::stop() noexcept {
     auto res = co_await task::stop();
+    _probe.clear();
     // task::stop() closed the runner's gate, so no run_impl is in flight and it
     // is safe to reset the state directly (unlike update_config, which races a
     // running fiber and defers via _config_changed). Reset so a later leader
@@ -203,30 +218,35 @@ bool mirroring_task::should_long_sync() const {
            >= full_sync_interval(_config);
 }
 
+model::schema_registry_sync_status
+mirroring_task::get_live_sync_status() const {
+    auto status = _status;
+    // Reflect the in-flight reconcile's live counters for mid-sync
+    // progress. Guarded on current_sync so it cannot double-count after the
+    // fold (which zeroes _reconcile_stats and bakes them into _status).
+    if (status.current_sync.has_value()) {
+        status.current_sync->summary.subject_versions_changed
+          += _reconcile_stats.versions_changed;
+        status.current_sync->summary.errors += _reconcile_stats.errors;
+        status.current_sync->summary.unsupported_features_removed
+          += _reconcile_stats.unsupported_features_removed;
+        status.totals_since_task_start.subject_versions_changed
+          += _reconcile_stats.versions_changed;
+        status.totals_since_task_start.errors += _reconcile_stats.errors;
+        status.totals_since_task_start.unsupported_features_removed
+          += _reconcile_stats.unsupported_features_removed;
+    }
+    return status;
+}
+
 model::task_status_report mirroring_task::get_status_report() const {
     auto report = task::get_status_report();
     // Only the shard leading _schemas/0 runs the sync; a stopped shard's empty
     // status must not win the admin aggregation over the leader's, so suppress
     // it.
     if (get_state() != model::task_state::stopped) {
-        auto status = _status;
-        // Reflect the in-flight reconcile's live counters for mid-sync
-        // progress. Guarded on current_sync so it cannot double-count after the
-        // fold (which zeroes _reconcile_stats and bakes them into _status).
-        if (status.current_sync.has_value()) {
-            status.current_sync->summary.subject_versions_changed
-              += _reconcile_stats.versions_changed;
-            status.current_sync->summary.errors += _reconcile_stats.errors;
-            status.current_sync->summary.unsupported_features_removed
-              += _reconcile_stats.unsupported_features_removed;
-            status.totals_since_task_start.subject_versions_changed
-              += _reconcile_stats.versions_changed;
-            status.totals_since_task_start.errors += _reconcile_stats.errors;
-            status.totals_since_task_start.unsupported_features_removed
-              += _reconcile_stats.unsupported_features_removed;
-        }
         report.detail = model::task_detail{
-          .schema_registry_sync_status = std::move(status)};
+          .schema_registry_sync_status = get_live_sync_status()};
     }
     return report;
 }

src/v/cluster_link/schema_registry_sync/mirroring_task.h

@@ -11,6 +11,7 @@
 
 #pragma once
 
+#include "cluster_link/schema_registry_sync/probe.h"
 #include "cluster_link/schema_registry_sync/reconciler.h"
 #include "cluster_link/schema_registry_sync/source_reader.h"
 #include "cluster_link/task.h"
@@ -73,6 +74,8 @@ class mirroring_task : public task {
 
     void update_config(const model::metadata& link_metadata) override;
 
+    ss::future<cl_result<void>> start() override;
+
     ss::future<cl_result<void>> stop() noexcept override;
 
     model::enabled_t is_enabled() const final;
@@ -224,6 +227,8 @@ class mirroring_task : public task {
     [[nodiscard]] state_transition make_active();
     [[nodiscard]] state_transition make_faulted(const ss::sstring& reason);
 
+    model::schema_registry_sync_status get_live_sync_status() const;
+
     model::schema_registry_sync_config _config;
     // Source->destination context remapping for the current run, rebuilt from
     // _config at the start of each full sync. Applied only at the destination
@@ -237,6 +242,7 @@ class mirroring_task : public task {
     // Live counters for the in-flight reconcile; reflected by get_status_report
     // for mid-sync progress, then folded into _status at end of run.
     reconcile_stats _reconcile_stats;
+    probe _probe;
     std::optional<ss::lowres_clock::time_point> _last_full_sync;
     // Set by update_config, consumed by run_impl to force a full scan. A flag
     // (rather than mutating _status/_last_full_sync in update_config) avoids

src/v/cluster_link/schema_registry_sync/probe.cc

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#include "cluster_link/schema_registry_sync/probe.h"
+
+#include "cluster_link/link_probe.h"
+
+#include <seastar/core/metrics.hh>
+
+namespace sm = ss::metrics;
+
+namespace cluster_link::schema_registry_sync {
+
+void probe::setup(const model::name_t& link_name, totals_fetcher get_totals) {
+    if (_metrics.has_value()) {
+        return;
+    }
+
+    const auto sl_name = link_probe::shadow_link_name(link_name);
+    _get_totals = std::move(get_totals);
+
+    _metrics.emplace().add_group(
+      link_probe::shadow_link_group,
+      {
+        sm::make_counter(
+          "schema_registry_subject_versions_changed",
+          [this] { return _get_totals().subject_versions_changed; },
+          sm::description(
+            "Number of subject versions created, updated or deleted on the "
+            "destination by Schema Registry shadowing since the task started"),
+          {sl_name})
+          .aggregate({sm::shard_label}),
+        sm::make_counter(
+          "schema_registry_compatibility_configs_changed",
+          [this] { return _get_totals().compatibility_configs_changed; },
+          sm::description(
+            "Number of compatibility configuration changes applied to the "
+            "destination by Schema Registry shadowing since the task started"),
+          {sl_name})
+          .aggregate({sm::shard_label}),
+        sm::make_counter(
+          "schema_registry_modes_changed",
+          [this] { return _get_totals().modes_changed; },
+          sm::description(
+            "Number of mode changes applied to the destination by Schema "
+            "Registry shadowing since the task started"),
+          {sl_name})
+          .aggregate({sm::shard_label}),
+        sm::make_counter(
+          "schema_registry_unsupported_features_removed",
+          [this] { return _get_totals().unsupported_features_removed; },
+          sm::description(
+            "Number of unsupported schema features removed from replicated "
+            "schemas by Schema Registry shadowing since the task started"),
+          {sl_name})
+          .aggregate({sm::shard_label}),
+        sm::make_counter(
+          "schema_registry_errors",
+          [this] { return _get_totals().errors; },
+          sm::description(
+            "Number of errors observed by Schema Registry "
+            "shadowing since the task started"),
+          {sl_name})
+          .aggregate({sm::shard_label}),
+      });
+}
+
+} // namespace cluster_link::schema_registry_sync

src/v/cluster_link/schema_registry_sync/probe.h

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#pragma once
+
+#include "cluster_link/model/types.h"
+#include "metrics/metrics.h"
+
+#include <seastar/util/noncopyable_function.hh>
+
+#include <optional>
+
+namespace cluster_link::schema_registry_sync {
+
+/// Exposes a Schema Registry shadowing task's totals_since_task_start
+/// counters on the internal and public prometheus endpoints.
+///
+/// The owning task registers the series when it starts leading
+/// `_schemas/0` and keeps them registered while paused (the totals
+/// survive a pause). They are removed when stop() completes: the task
+/// state flips to `stopped` before the runner drains, so a scrape
+/// during that window still sees the series. Stopping resets the
+/// totals, so a stale series cannot linger after leadership moves
+/// away; a transfer may briefly export the series on zero or two nodes.
+class probe {
+public:
+    using totals_fetcher
+      = ss::noncopyable_function<model::schema_registry_sync_summary()>;
+
+    probe() = default;
+    probe(const probe&) = delete;
+    probe& operator=(const probe&) = delete;
+    probe(probe&&) = delete;
+    probe& operator=(probe&&) = delete;
+    ~probe() = default;
+
+    /// Registers the counter series for `link_name`, fetching the current
+    /// totals through `get_totals` on every scrape. Idempotent: resuming a
+    /// paused task invokes it again.
+    void setup(const model::name_t& link_name, totals_fetcher get_totals);
+
+    /// Removes the registered series.
+    void clear() { _metrics.reset(); }
+
+private:
+    totals_fetcher _get_totals;
+    std::optional<metrics::all_metrics_groups> _metrics;
+};
+
+} // namespace cluster_link::schema_registry_sync

src/v/cluster_link/schema_registry_sync/tests/BUILD

@@ -18,11 +18,13 @@ redpanda_cc_gtest(
         "//src/v/cluster_link/tests:test_deps",
         "//src/v/container:chunked_hash_map",
         "//src/v/container:chunked_vector",
+        "//src/v/metrics",
         "//src/v/model",
         "//src/v/pandaproxy/schema_registry:types",
         "//src/v/schema:registry",
         "//src/v/schema/tests:fake_registry",
         "//src/v/test_utils:gtest",
+        "//src/v/test_utils:metrics",
         "@googletest//:gtest",
         "@seastar",
     ],