From 7c3c543f66c63de847fb31425c7c4f916a5e445a Mon Sep 17 00:00:00 2001 From: Andrew Nguyen Date: Thu, 2 Jul 2026 04:59:32 +0000 Subject: [PATCH 1/3] cluster_link: export Schema Registry sync totals as metrics Expose the totals_since_task_start counters of the Schema Registry shadowing task as counters on both the internal and public metrics endpoints, labelled by shadow_link_name. The task drives the series off its state machine: they exist exactly while it surfaces Schema Registry status (any state but stopped), so a demoted leader stops exporting rather than double-counting into the aggregated sum (a transfer may briefly export on zero or two nodes), while pause/resume keeps the series continuous. stop() already resets the totals, so a fresh leader re-exports from zero. --- src/v/cluster_link/schema_registry_sync/BUILD | 3 + .../schema_registry_sync/mirroring_task.cc | 54 ++++++++----- .../schema_registry_sync/mirroring_task.h | 6 ++ .../schema_registry_sync/probe.cc | 76 +++++++++++++++++++ .../cluster_link/schema_registry_sync/probe.h | 58 ++++++++++++++ 5 files changed, 180 insertions(+), 17 deletions(-) create mode 100644 src/v/cluster_link/schema_registry_sync/probe.cc create mode 100644 src/v/cluster_link/schema_registry_sync/probe.h diff --git a/src/v/cluster_link/schema_registry_sync/BUILD b/src/v/cluster_link/schema_registry_sync/BUILD index 5c50bd81c8b9a..51be43e1439b8 100644 --- a/src/v/cluster_link/schema_registry_sync/BUILD +++ b/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", diff --git a/src/v/cluster_link/schema_registry_sync/mirroring_task.cc b/src/v/cluster_link/schema_registry_sync/mirroring_task.cc index 7cb865b8d60a4..dbdcc01c02e98 100644 --- a/src/v/cluster_link/schema_registry_sync/mirroring_task.cc +++ b/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> 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> 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; } diff --git a/src/v/cluster_link/schema_registry_sync/mirroring_task.h b/src/v/cluster_link/schema_registry_sync/mirroring_task.h index 0fda87dfa7a09..1ee8b8e95be87 100644 --- a/src/v/cluster_link/schema_registry_sync/mirroring_task.h +++ b/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> start() override; + ss::future> 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 _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 diff --git a/src/v/cluster_link/schema_registry_sync/probe.cc b/src/v/cluster_link/schema_registry_sync/probe.cc new file mode 100644 index 0000000000000..3c9d4631431e6 --- /dev/null +++ b/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 + +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 diff --git a/src/v/cluster_link/schema_registry_sync/probe.h b/src/v/cluster_link/schema_registry_sync/probe.h new file mode 100644 index 0000000000000..7e6cbab761eab --- /dev/null +++ b/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 + +#include + +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; + + 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; +}; + +} // namespace cluster_link::schema_registry_sync From 26e9f1c1ee30c798dede40676fea5b614e3821ac Mon Sep 17 00:00:00 2001 From: Andrew Nguyen Date: Fri, 10 Jul 2026 01:06:46 +0000 Subject: [PATCH 2/3] cluster_link: unit-test SR-sync metric export Assert the probe's counters mirror totals_since_task_start on both the internal and public registries, labelled with the link name, and pin the series lifecycle: registered while active, kept (with values) across pause/resume, removed on stop so a non-leader cannot export misleading reset zeros. --- .../schema_registry_sync/tests/BUILD | 2 + .../tests/mirroring_task_test.cc | 287 ++++++++++++++++++ 2 files changed, 289 insertions(+) diff --git a/src/v/cluster_link/schema_registry_sync/tests/BUILD b/src/v/cluster_link/schema_registry_sync/tests/BUILD index df6454172c59c..b97f64e8c3b25 100644 --- a/src/v/cluster_link/schema_registry_sync/tests/BUILD +++ b/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", ], diff --git a/src/v/cluster_link/schema_registry_sync/tests/mirroring_task_test.cc b/src/v/cluster_link/schema_registry_sync/tests/mirroring_task_test.cc index 090ead2068bf9..1da8d68d72467 100644 --- a/src/v/cluster_link/schema_registry_sync/tests/mirroring_task_test.cc +++ b/src/v/cluster_link/schema_registry_sync/tests/mirroring_task_test.cc @@ -9,15 +9,19 @@ * by the Apache License, Version 2.0 */ +#include "cluster_link/link_probe.h" #include "cluster_link/schema_registry_sync/mirroring_task.h" +#include "cluster_link/schema_registry_sync/probe.h" #include "cluster_link/schema_registry_sync/source_reader.h" #include "cluster_link/schema_registry_sync/tests/sr_sync_test_fixtures.h" #include "cluster_link/tests/deps.h" #include "container/chunked_vector.h" +#include "metrics/metrics.h" #include "model/namespace.h" #include "pandaproxy/schema_registry/types.h" #include "schema/tests/fake_registry.h" #include "test_utils/async.h" +#include "test_utils/metrics.h" #include "test_utils/test.h" #include @@ -25,6 +29,8 @@ #include +#include + using namespace std::chrono_literals; namespace cluster_link::tests { @@ -35,6 +41,21 @@ static const model::name_t link_name{"test_sr_link"}; constexpr auto tail_interval = 1s; constexpr auto wait_interval = 5s; +// Reads one of the probe's counter series for the test link on the current +// shard (the shard leading `_schemas/0` in these tests). `counter` is the +// name suffix after "schema_registry_"; `handle` selects the internal or +// public registry. nullopt when the series is not registered. +std::optional sr_sync_metric(std::string_view counter, int handle) { + return test_utils::find_metric_value( + fmt::format( + "{}_schema_registry_{}", link_probe::shadow_link_group, counter), + handle, + {{link_probe::shadow_link_name.name(), link_name()}}); +} + +const auto both_metric_handles = std::to_array( + {ss::metrics::default_handle(), metrics::public_metrics_handle}); + model::metadata get_default_metadata() { model::metadata metadata{ .name = link_name, @@ -1238,6 +1259,272 @@ TEST_F(mirroring_task_test, follows_partition_leadership) { EXPECT_FALSE(task->detail.has_value()); } +TEST_F(mirroring_task_test, exports_sync_totals_on_both_metric_endpoints) { + // Three subject versions imported by the first full sync; the probe's + // counters must mirror totals_since_task_start on both the internal and + // the public registry, labelled with the link name. + _source_state.add(ppsr::context_subject::unqualified("a"), 1); + _source_state.add(ppsr::context_subject::unqualified("b"), 1); + _source_state.add(ppsr::context_subject::unqualified("c"), 1); + + lead_schema_registry(); + fixture()->upsert_link(get_default_metadata()).get(); + + auto status = wait_for_sync_status([](const auto& s) { + return s.last_full_sync.has_value() + && !s.current_sync.has_value(); + }).get(); + ASSERT_TRUE(status.has_value()); + ASSERT_EQ(status->totals_since_task_start.subject_versions_changed, 3); + + for (auto handle : both_metric_handles) { + EXPECT_EQ( + sr_sync_metric("subject_versions_changed", handle), + std::optional{3}); + // The other counters have not moved, but their series exist. + EXPECT_EQ( + sr_sync_metric("compatibility_configs_changed", handle), + std::optional{0}); + EXPECT_EQ( + sr_sync_metric("modes_changed", handle), std::optional{0}); + EXPECT_EQ( + sr_sync_metric("unsupported_features_removed", handle), + std::optional{0}); + EXPECT_EQ(sr_sync_metric("errors", handle), std::optional{0}); + } +} + +TEST_F(mirroring_task_test, metric_series_survive_pause_and_drop_on_stop) { + _source_state.add(ppsr::context_subject::unqualified("orders-value"), 1); + + lead_schema_registry(); + fixture()->upsert_link(get_default_metadata()).get(); + auto status = wait_for_sync_status([](const auto& s) { + return s.totals_since_task_start.subject_versions_changed + == 1; + }).get(); + ASSERT_TRUE(status.has_value()); + + // Pausing (config-disabled while still leading) keeps the task's totals, + // so the series must stay registered and keep their values. + auto paused = get_default_metadata(); + paused.configuration.schema_registry_sync_cfg.api_mode()->is_enabled + = model::enabled_t::no; + fixture()->update_link(model::id_t{0}, std::move(paused)).get(); + ASSERT_TRUE(wait_for_task_state(model::task_state::paused).get()); + for (auto handle : both_metric_handles) { + EXPECT_EQ( + sr_sync_metric("subject_versions_changed", handle), + std::optional{1}); + } + + // Resuming re-enters start(), whose probe setup must be idempotent. + fixture()->update_link(model::id_t{0}, get_default_metadata()).get(); + ASSERT_TRUE(wait_for_task_state(model::task_state::active).get()); + for (auto handle : both_metric_handles) { + EXPECT_EQ( + sr_sync_metric("subject_versions_changed", handle), + std::optional{1}); + } + + // Losing `_schemas/0` leadership stops the task; a stopped task's totals + // are reset, so a lingering series would export misleading zeros -- the + // series must be removed outright. The state flips to stopped before the + // runner drains and the probe clears, so poll rather than assert. + unlead_schema_registry(); + ASSERT_TRUE(wait_for_task_state(model::task_state::stopped).get()); + for (auto handle : both_metric_handles) { + ::tests::cooperative_spin_wait_with_timeout(wait_interval, [handle] { + return !sr_sync_metric("subject_versions_changed", handle) + .has_value(); + }).get(); + } +} + +TEST_F(mirroring_task_test, totals_reset_across_leadership_tenures) { + // stop() resets the sync state after dropping the series; a task that + // regains `_schemas/0` leadership (A->B->A) re-registers the series and + // must export the new tenure's totals only. A scrape after the reset must + // read the reassigned _status, not anything bound at first setup. + auto subject = ppsr::context_subject::unqualified("orders-value"); + _source_state.add(subject, 1); + + lead_schema_registry(); + fixture()->upsert_link(get_default_metadata()).get(); + auto first = wait_for_sync_status([](const auto& s) { + return s.totals_since_task_start.subject_versions_changed + == 1; + }).get(); + ASSERT_TRUE(first.has_value()); + + unlead_schema_registry(); + ASSERT_TRUE(wait_for_task_state(model::task_state::stopped).get()); + + // v2 appears while not leading; the second tenure's first full sync + // imports only the missing version, so its totals are exactly 1. + _source_state.add(subject, 2); + lead_schema_registry(); + auto second = wait_for_sync_status([](const auto& s) { + return s.last_full_sync.has_value() + && !s.current_sync.has_value(); + }).get(); + ASSERT_TRUE(second.has_value()); + EXPECT_EQ(second->totals_since_task_start.subject_versions_changed, 1); + + // 2 would mean the first tenure's totals leaked through the reset. + for (auto handle : both_metric_handles) { + EXPECT_EQ( + sr_sync_metric("subject_versions_changed", handle), + std::optional{1}); + } +} + +TEST_F(mirroring_task_test, metric_values_are_fetched_live_on_scrape) { + model::schema_registry_sync_summary totals; + srs::probe probe; + probe.setup(link_name, [&totals] { return totals; }); + + const auto counters = std::to_array>( + {{"subject_versions_changed", 1}, + {"compatibility_configs_changed", 2}, + {"modes_changed", 3}, + {"unsupported_features_removed", 4}, + {"errors", 5}}); + + for (auto handle : both_metric_handles) { + for (const auto& counter : counters) { + EXPECT_EQ( + sr_sync_metric(counter.first, handle), + std::optional{0}); + } + } + + totals.subject_versions_changed = 1; + totals.compatibility_configs_changed = 2; + totals.modes_changed = 3; + totals.unsupported_features_removed = 4; + totals.errors = 5; + for (auto handle : both_metric_handles) { + for (const auto& [name, value] : counters) { + EXPECT_EQ( + sr_sync_metric(name, handle), std::optional{value}); + } + } + + probe.clear(); +} + +// Fixture whose destination parks an import mid-reconcile, so a test can +// observe the task while _reconcile_stats holds counts not yet folded into +// _status. +class mirroring_task_blocking_import_test : public mirroring_task_test { +public: + // A parked import holds the runner's gate; the task stop in the base + // teardown would wait on it forever, so release first. + ss::future<> TearDownAsync() override { + release(); + co_await mirroring_task_test::TearDownAsync(); + } + +protected: + schema::registry* dest() override { return &_blocking; } + + void release() { + if (!_released) { + _released = true; + _blocking.unblock(); + } + } + + blocking_import_registry _blocking{&_registry, /*block_after=*/1}; + +private: + bool _released{false}; +}; + +TEST_F( + mirroring_task_blocking_import_test, + metrics_include_in_flight_reconcile_stats) { + // The reconciler counts each import in _reconcile_stats, which is folded + // into _status only at end of run. Parking the second import mid-reconcile + // pins that the exported counters include the in-flight stats. Only + // EXPECTs between park and release, so the parked import is always + // released. + _source_state.add(ppsr::context_subject::unqualified("a"), 1); + _source_state.add(ppsr::context_subject::unqualified("b"), 1); + + lead_schema_registry(); + fixture()->upsert_link(get_default_metadata()).get(); + + _blocking.entered().get(); + // entered() races the forwarded import's completion continuation; poll + // the report (which shares the live fold) until the count shows. + auto status + = wait_for_sync_status([](const auto& s) { + return s.current_sync.has_value() + && s.totals_since_task_start.subject_versions_changed == 1; + }).get(); + EXPECT_TRUE(status.has_value()); + + for (auto handle : both_metric_handles) { + EXPECT_EQ( + sr_sync_metric("subject_versions_changed", handle), + std::optional{1}); + } + + release(); + auto done = wait_for_sync_status([](const auto& s) { + return s.last_full_sync.has_value() + && !s.current_sync.has_value(); + }).get(); + EXPECT_TRUE(done.has_value()); + for (auto handle : both_metric_handles) { + EXPECT_EQ( + sr_sync_metric("subject_versions_changed", handle), + std::optional{2}); + } +} + +TEST_F( + mirroring_task_blocking_import_test, + scrape_during_stop_drain_exports_last_totals) { + // Losing leadership flips the task to stopped immediately, but the series + // are only dropped once the runner's gate drains; a parked import wedges + // stop() in exactly that window. A scrape landing there must still export + // the last live totals -- reading them through the status report instead + // would hit the optional that get_status_report() leaves disengaged while + // stopped. + _source_state.add(ppsr::context_subject::unqualified("a"), 1); + _source_state.add(ppsr::context_subject::unqualified("b"), 1); + + lead_schema_registry(); + fixture()->upsert_link(get_default_metadata()).get(); + + _blocking.entered().get(); + auto status + = wait_for_sync_status([](const auto& s) { + return s.current_sync.has_value() + && s.totals_since_task_start.subject_versions_changed == 1; + }).get(); + EXPECT_TRUE(status.has_value()); + + unlead_schema_registry(); + ASSERT_TRUE(wait_for_task_state(model::task_state::stopped).get()); + for (auto handle : both_metric_handles) { + EXPECT_EQ( + sr_sync_metric("subject_versions_changed", handle), + std::optional{1}); + } + + release(); + // The drain completes, stop() drops the series and resets the totals. + ::tests::cooperative_spin_wait_with_timeout(wait_interval, [] { + return !sr_sync_metric( + "subject_versions_changed", ss::metrics::default_handle()) + .has_value(); + }).get(); +} + TEST_F(mirroring_task_test, destination_inventory_spans_contexts_and_deleted) { auto a = ppsr::context_subject::unqualified("a"); auto c = ppsr::context_subject{ppsr::context{".b"}, ppsr::subject{"c"}}; From 97d7209312d91c3206ce04b4ca26c42e09a98792 Mon Sep 17 00:00:00 2001 From: Andrew Nguyen Date: Fri, 10 Jul 2026 01:06:54 +0000 Subject: [PATCH 3/3] cluster_link: e2e-test SR-sync prometheus counters Scrape the shadow_link_schema_registry_* counters in the existing SR sync tests: the e2e counter check now asserts both metrics endpoints agree with the admin API totals, the out-of-scope-reference test covers a nonzero errors counter, and the leadership-change test verifies the series follows _schemas/0 leadership (the old leader drops its series rather than exporting stale values). Disable the leader balancer for the suite: totals_since_task_start resets when _schemas/0 leadership moves, so a balancer-initiated move would strand the exact-value counter assertions. The leadership-change test moves leadership explicitly. --- ...uster_linking_schema_registry_sync_test.py | 111 +++++++++++++++++- 1 file changed, 109 insertions(+), 2 deletions(-) diff --git a/tests/rptest/tests/cluster_linking_schema_registry_sync_test.py b/tests/rptest/tests/cluster_linking_schema_registry_sync_test.py index 46578d4d85b6f..92bc6262a95d5 100644 --- a/tests/rptest/tests/cluster_linking_schema_registry_sync_test.py +++ b/tests/rptest/tests/cluster_linking_schema_registry_sync_test.py @@ -33,7 +33,7 @@ SecondaryClusterSpec, ServiceType, ) -from rptest.services.redpanda import SchemaRegistryConfig +from rptest.services.redpanda import MetricsEndpoint, SchemaRegistryConfig from rptest.tests.cluster_linking_test_base import ShadowLinkTestBase from rptest.tests.schema_registry_test import SchemaRegistryRedpandaClient from rptest.util import firewall_blocked @@ -351,6 +351,62 @@ def synced() -> bool: # --- status-counter checks --------------------------------------------- + SR_SYNC_METRIC_COUNTERS = ( + "subject_versions_changed", + "compatibility_configs_changed", + "modes_changed", + "unsupported_features_removed", + "errors", + ) + + def _sr_sync_metric_totals( + self, metrics_endpoint: MetricsEndpoint + ) -> dict[str, float] | None: + """Cluster-wide sums of the shadow-link Schema Registry counters for + LINK_NAME from the given metrics endpoint, keyed by counter name, or + None while any counter has no series (e.g. no leader yet).""" + patterns = { + counter: f"shadow_link_schema_registry_{counter}" + for counter in self.SR_SYNC_METRIC_COUNTERS + } + samples = self.target_cluster_service.metrics_samples( + sample_patterns=list(patterns.values()), metrics_endpoint=metrics_endpoint + ) + totals: dict[str, float] = {} + for counter, pattern in patterns.items(): + if pattern not in samples: + return None + values = [ + s.value + for s in samples[pattern].samples + if s.labels["shadow_link_name"] == LINK_NAME + ] + if not values: + return None + totals[counter] = sum(values) + return totals + + def _versions_changed_metric(self, metrics_endpoint: MetricsEndpoint) -> float: + """The cluster-wide subject_versions_changed counter, or -1 while the + series is absent (distinguishable from a real 0 so wait_until + predicates can wait for either state).""" + totals = self._sr_sync_metric_totals(metrics_endpoint) + return totals["subject_versions_changed"] if totals is not None else -1.0 + + def _verify_prometheus_counters(self, expected_versions: int): + # Both metrics endpoints must agree with the admin API's + # totals_since_task_start. The counters read the task's live status at + # scrape time, so once the admin totals have converged no extra wait is + # needed. + for endpoint in (MetricsEndpoint.METRICS, MetricsEndpoint.PUBLIC_METRICS): + totals = self._sr_sync_metric_totals(endpoint) + assert totals is not None, f"no SR sync series on {endpoint}" + self.logger.info(f"[{endpoint}] SR sync counters: {totals}") + assert totals["subject_versions_changed"] == expected_versions, totals + assert totals["errors"] == 0, totals + for name in self.EXPECTED_ZERO_COUNTERS: + assert totals[name] == 0, totals + # Counters expected to stay zero in the override-free DAG tests: those DAGs # register no mode or compatibility overrides and carry no unsupported # fields, so mode/config replication and unsupported-feature handling are @@ -441,7 +497,9 @@ def counters_ready() -> bool: f"unexpected {name}={getattr(totals, name)}" ) - # Cross-check the rpk `shadow status` rendering against the admin API. + # Cross-check the prometheus counters and the rpk `shadow status` + # rendering against the admin API. + self._verify_prometheus_counters(expected_versions) self._verify_rpk_status(expected_subjects, expected_versions) def _log_counters(self, source: str, sr): @@ -900,6 +958,14 @@ def errored() -> bool: err_msg="out-of-scope reference did not surface as a counted error", ) + # The counted error is also exported through the errors counter on + # both prometheus endpoints (the value tests above only ever see it at + # zero). + for endpoint in (MetricsEndpoint.METRICS, MetricsEndpoint.PUBLIC_METRICS): + totals = self._sr_sync_metric_totals(endpoint) + assert totals is not None, f"no SR sync series on {endpoint}" + assert totals["errors"] >= 1, totals + # The referrer never imported; the in-scope independent subject did. dest_subjects = set(dest.get_subjects().json()) assert "ok-value" in dest_subjects, dest_subjects @@ -1023,6 +1089,16 @@ def first_synced() -> bool: before_start = ( self._admin_sr_status().totals_since_task_start.start_time.ToNanoseconds() ) + # The first instance exports its totals as prometheus counters from the + # leader broker (summed cluster-wide; only the leader has the series). + before_changed = ( + self._admin_sr_status().totals_since_task_start.subject_versions_changed + ) + assert before_changed >= expected_versions + assert ( + self._versions_changed_metric(MetricsEndpoint.PUBLIC_METRICS) + == before_changed + ) # Move _schemas/0 leadership to another destination broker. The sync # task follows leadership, so a fresh instance takes over there. @@ -1073,6 +1149,17 @@ def re_derived() -> bool: err_msg="new leader did not re-derive counters after the bounce", ) + # The cluster-wide metric drops to the new instance's 0: the old + # leader unregistered its series on stop (a lingering stale series + # would keep the sum at the pre-bounce value), and the new leader + # exports the reset totals. -1 (no series anywhere) must not pass. + wait_until( + lambda: self._versions_changed_metric(MetricsEndpoint.PUBLIC_METRICS) == 0, + timeout_sec=30, + backoff_sec=1, + err_msg="prometheus counter did not follow leadership to the new broker", + ) + # The new leader keeps syncing: a subject added now is imported by it, # advancing the new instance's cumulative change counter. base_changed = ( @@ -1136,6 +1223,15 @@ def returning_leader_fresh() -> bool: err_msg="reused instance did not reset state on regaining leadership", ) + # Same series check for the return leg: B unregistered on stop and the + # reused instance on A re-registered with its reset totals. + wait_until( + lambda: self._versions_changed_metric(MetricsEndpoint.PUBLIC_METRICS) == 0, + timeout_sec=30, + backoff_sec=1, + err_msg="prometheus counter did not follow leadership back to the original broker", + ) + # --- unsupported-feature policy ---------------------------------------- # Confluent-only: only a Confluent source accepts the unsupported fields # (schema metadata.tags, config compatibilityGroup) on registration, so a @@ -1283,6 +1379,14 @@ def __init__(self, test_context: TestContext, *args: Any, **kwargs: Any): ), # Destination (primary) cluster Schema Registry. schema_registry_config=SchemaRegistryConfig(), + # The sync task runs on whichever destination broker leads + # _schemas/0, and totals_since_task_start (admin counters and the + # prometheus series) resets whenever leadership moves. With the + # destination fully synced a fresh instance imports nothing, so a + # balancer-initiated move would strand every exact-value counter + # wait/assert at 0. Leadership movement is exercised explicitly by + # test_schema_registry_api_sync_survives_leadership_change. + extra_rp_conf={"enable_leader_balancer": False}, *args, **kwargs, ) @@ -1361,6 +1465,9 @@ def __init__(self, test_context: TestContext, *args: Any, **kwargs: Any): # + 1 kafka + 1 SR). A one-broker source is fine here -- the SR's # _schemas topic is RF=1. secondary_cluster_args=SecondaryClusterArgs(num_brokers=1), + # Keep _schemas/0 leadership stable for the exact-value counter + # asserts; same rationale as SchemaRegistrySyncE2ETest. + extra_rp_conf={"enable_leader_balancer": False}, *args, **kwargs, )