diff --git a/src/v/cloud_topics/level_one/metastore/lsm/BUILD b/src/v/cloud_topics/level_one/metastore/lsm/BUILD index f09971700af2b..52bfbcf9503be 100644 --- a/src/v/cloud_topics/level_one/metastore/lsm/BUILD +++ b/src/v/cloud_topics/level_one/metastore/lsm/BUILD @@ -212,6 +212,28 @@ redpanda_cc_library( ], ) +redpanda_cc_library( + name = "metastore_lsm_probe", + srcs = [ + "metastore_lsm_probe.cc", + ], + hdrs = [ + "metastore_lsm_probe.h", + ], + implementation_deps = [ + "//src/v/config", + "//src/v/ssx:sformat", + ], + visibility = ["//visibility:public"], + deps = [ + "//src/v/base", + "//src/v/lsm", + "//src/v/metrics", + "//src/v/model", + "@seastar", + ], +) + redpanda_cc_library( name = "replicated_db", srcs = [ @@ -227,6 +249,7 @@ redpanda_cc_library( "//src/v/config", "//src/v/lsm/io:cloud_cache_persistence", "//src/v/model:batch_builder", + "//src/v/raft", "//src/v/serde", "//src/v/ssx:clock", "//src/v/ssx:time", @@ -234,6 +257,7 @@ redpanda_cc_library( visibility = ["//visibility:public"], deps = [ ":lsm_update", + ":metastore_lsm_probe", ":stm", ":write_batch_row", "//src/v/cloud_io:cache", diff --git a/src/v/cloud_topics/level_one/metastore/lsm/metastore_lsm_probe.cc b/src/v/cloud_topics/level_one/metastore/lsm/metastore_lsm_probe.cc new file mode 100644 index 0000000000000..494925a2b1bec --- /dev/null +++ b/src/v/cloud_topics/level_one/metastore/lsm/metastore_lsm_probe.cc @@ -0,0 +1,102 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Licensed as a Redpanda Enterprise file under the Redpanda Community + * License (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md + */ +#include "cloud_topics/level_one/metastore/lsm/metastore_lsm_probe.h" + +#include "config/configuration.h" +#include "metrics/prometheus_sanitize.h" +#include "ssx/sformat.h" + +#include + +namespace cloud_topics::l1 { + +metastore_lsm_probe::metastore_lsm_probe( + model::partition_id metastore_partition) + : _probe(ss::make_lw_shared()) { + if (config::shard_local_cfg().disable_metrics()) { + return; + } + + namespace sm = ss::metrics; + const std::vector labels = { + sm::label("metastore_partition")( + ssx::sformat("{}", metastore_partition())), + }; + + _metrics.add_group( + prometheus_sanitize::metrics_name("cloud_topics:l1:metastore_lsm"), + { + sm::make_counter( + "block_cache_hits_total", + [this] { return _probe->block_cache_hit; }, + sm::description( + "Number of LSM block cache hits for the L1 metastore database."), + labels), + sm::make_counter( + "block_cache_misses_total", + [this] { return _probe->block_cache_miss; }, + sm::description( + "Number of LSM block cache misses for the L1 metastore database. " + "A high miss rate against a working set larger than " + "cloud_topics_metastore_block_cache_size indicates the cache is " + "undersized for the workload."), + labels), + sm::make_counter( + "table_cache_hits_total", + [this] { return _probe->table_cache_hit; }, + sm::description( + "Number of LSM table (SST file handle) cache hits for the L1 " + "metastore database."), + labels), + sm::make_counter( + "table_cache_misses_total", + [this] { return _probe->table_cache_miss; }, + sm::description( + "Number of LSM table cache misses for the L1 metastore database. " + "Each miss opens an SST file."), + labels), + sm::make_counter( + "throttled_writes_total", + [this] { return _probe->throttled_writes; }, + sm::description( + "Number of LSM writes that were throttled because L0 reached " + "level_zero_slowdown_writes_trigger."), + labels), + sm::make_counter( + "stalled_writes_total", + [this] { return _probe->stalled_writes; }, + sm::description( + "Number of LSM writes that were stalled because L0 reached " + "level_zero_stop_writes_trigger."), + labels), + sm::make_histogram( + "compaction_duration_microseconds", + [this] { + return _probe->compaction_latency.internal_histogram_logform(); + }, + sm::description("Duration of LSM compaction runs."), + labels), + sm::make_histogram( + "flush_duration_microseconds", + [this] { return _probe->flush_latency.internal_histogram_logform(); }, + sm::description("Duration of LSM memtable flushes."), + labels), + sm::make_histogram( + "manifest_write_duration_microseconds", + [this] { + return _probe->manifest_write_latency + .internal_histogram_logform(); + }, + sm::description("Duration of LSM manifest writes."), + labels), + }); +} + +} // namespace cloud_topics::l1 diff --git a/src/v/cloud_topics/level_one/metastore/lsm/metastore_lsm_probe.h b/src/v/cloud_topics/level_one/metastore/lsm/metastore_lsm_probe.h new file mode 100644 index 0000000000000..5c68ab844f934 --- /dev/null +++ b/src/v/cloud_topics/level_one/metastore/lsm/metastore_lsm_probe.h @@ -0,0 +1,52 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Licensed as a Redpanda Enterprise file under the Redpanda Community + * License (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md + */ +#pragma once + +#include "base/seastarx.h" +#include "lsm/lsm.h" +#include "metrics/metrics.h" +#include "model/fundamental.h" + +#include + +namespace cloud_topics::l1 { + +/// Per-metastore-partition wrapper around lsm::probe that registers the +/// underlying counters and histograms with the seastar metrics system. +/// +/// One instance per replicated_database; lifetime matches the database, so +/// leadership churn re-registers the metric set. +class metastore_lsm_probe { +public: + explicit metastore_lsm_probe(model::partition_id metastore_partition); + + metastore_lsm_probe(const metastore_lsm_probe&) = delete; + metastore_lsm_probe& operator=(const metastore_lsm_probe&) = delete; + metastore_lsm_probe(metastore_lsm_probe&&) = delete; + metastore_lsm_probe& operator=(metastore_lsm_probe&&) = delete; + ~metastore_lsm_probe() = default; + + /// The shared probe pointer to pass to lsm::options.probe. + ss::lw_shared_ptr probe() const { return _probe; } + + /// Eagerly unregister the metric group. Callers should invoke this at the + /// well-defined teardown point (e.g. replicated_database::close()) rather + /// than relying on the destructor, so a new probe for the same + /// metastore_partition can be registered while the old shared_ptr to the + /// owning replicated_database is still alive (held by an in-flight RPC + /// fiber). + void deregister_metrics() { _metrics.clear(); } + +private: + ss::lw_shared_ptr _probe; + metrics::internal_metric_groups _metrics; +}; + +} // namespace cloud_topics::l1 diff --git a/src/v/cloud_topics/level_one/metastore/lsm/replicated_db.cc b/src/v/cloud_topics/level_one/metastore/lsm/replicated_db.cc index fee20851d6958..c42fbecdb58b4 100644 --- a/src/v/cloud_topics/level_one/metastore/lsm/replicated_db.cc +++ b/src/v/cloud_topics/level_one/metastore/lsm/replicated_db.cc @@ -12,6 +12,7 @@ #include "cloud_topics/level_one/metastore/domain_uuid.h" #include "cloud_topics/level_one/metastore/lsm/lsm_update.h" +#include "cloud_topics/level_one/metastore/lsm/metastore_lsm_probe.h" #include "cloud_topics/level_one/metastore/lsm/replicated_persistence.h" #include "cloud_topics/level_one/metastore/lsm/stm.h" #include "cloud_topics/logger.h" @@ -21,6 +22,7 @@ #include "lsm/proto/manifest.proto.h" #include "model/batch_builder.h" #include "model/record.h" +#include "raft/consensus.h" #include "serde/rw/scalar.h" #include "ssx/clock.h" #include "ssx/future-util.h" @@ -138,6 +140,13 @@ replicated_database::open( .metadata = std::move(meta_persist_fut.get()), }; + // Construct and register the per-metastore-partition LSM probe before + // opening the database, so the LSM's counters and histograms are wired + // into the seastar metrics system from the first operation. Labeled by + // the metastore_partition for which this replica is the leader. + auto lsm_probe = std::make_unique( + s->raft()->ntp().tp.partition); + // Open the LSM database using the persisted manifest from the STM. auto db_fut = co_await ss::coroutine::as_future( lsm::database::open( @@ -154,6 +163,7 @@ replicated_database::open( .file_deletion_delay = absl::FromChrono( config::shard_local_cfg() .cloud_topics_long_term_file_deletion_delay()), + .probe = lsm_probe->probe(), }, std::move(io))); if (db_fut.failed()) { @@ -201,8 +211,8 @@ replicated_database::open( } } } - auto ret = std::unique_ptr( - new replicated_database(term, domain_uuid, s, std::move(db), as, sg)); + auto ret = std::unique_ptr(new replicated_database( + term, domain_uuid, s, std::move(db), std::move(lsm_probe), as, sg)); ret->start(); co_return std::move(ret); } @@ -224,6 +234,11 @@ replicated_database::close() { wrap_failed_future(fut.get_exception(), "Error closing database")); } co_await std::move(gate_fut); + // Tear down the probe's metric registration eagerly so a new + // replicated_database for the same metastore_partition can register + // its own probe even if some in-flight reference keeps this object + // alive past close(). + probe_->deregister_metrics(); co_return std::expected{}; } diff --git a/src/v/cloud_topics/level_one/metastore/lsm/replicated_db.h b/src/v/cloud_topics/level_one/metastore/lsm/replicated_db.h index df21fb6a128b3..9f7f1131a0b47 100644 --- a/src/v/cloud_topics/level_one/metastore/lsm/replicated_db.h +++ b/src/v/cloud_topics/level_one/metastore/lsm/replicated_db.h @@ -12,6 +12,7 @@ #include "cloud_io/cache_service.h" #include "cloud_io/remote.h" #include "cloud_storage_clients/types.h" +#include "cloud_topics/level_one/metastore/lsm/metastore_lsm_probe.h" #include "cloud_topics/level_one/metastore/lsm/stm.h" #include "cloud_topics/level_one/metastore/lsm/write_batch_row.h" #include "container/chunked_vector.h" @@ -25,6 +26,7 @@ #include #include +#include namespace cloud_topics::l1 { @@ -103,12 +105,14 @@ class replicated_database { domain_uuid domain_uuid, stm* s, lsm::database db, + std::unique_ptr probe, ss::abort_source& as, ss::scheduling_group sg) : term_(term) , expected_domain_uuid_(domain_uuid) , stm_(s) , db_(std::move(db)) + , probe_(std::move(probe)) , as_(as) , sg_(sg) {} @@ -137,6 +141,10 @@ class replicated_database { // The underlying LSM database. lsm::database db_; + // Wraps and exposes db_'s lsm::probe via seastar metrics. Outlives db_ + // closure logically but tied to the same instance lifetime. + std::unique_ptr probe_; + ss::gate gate_; ss::abort_source& as_; ss::scheduling_group sg_;