ct/l1/metastore: expose LSM probe metrics

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,13 +249,15 @@ 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",
     ],
     visibility = ["//visibility:public"],
     deps = [
         ":lsm_update",
+        ":metastore_lsm_probe",
         ":stm",
         ":write_batch_row",
         "//src/v/cloud_io:cache",

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 <seastar/core/metrics.hh>
+
+namespace cloud_topics::l1 {
+
+metastore_lsm_probe::metastore_lsm_probe(
+  model::partition_id metastore_partition)
+  : _probe(ss::make_lw_shared<lsm::probe>()) {
+    if (config::shard_local_cfg().disable_metrics()) {
+        return;
+    }
+
+    namespace sm = ss::metrics;
+    const std::vector<sm::label_instance> 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

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 <seastar/core/shared_ptr.hh>
+
+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<lsm::probe> 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<lsm::probe> _probe;
+    metrics::internal_metric_groups _metrics;
+};
+
+} // namespace cloud_topics::l1

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<metastore_lsm_probe>(
+      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<replicated_database>(
-      new replicated_database(term, domain_uuid, s, std::move(db), as, sg));
+    auto ret = std::unique_ptr<replicated_database>(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<void, error>{};
 }
 

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 <expected>
 #include <filesystem>
+#include <memory>
 
 namespace cloud_topics::l1 {
 
@@ -103,12 +105,14 @@ class replicated_database {
       domain_uuid domain_uuid,
       stm* s,
       lsm::database db,
+      std::unique_ptr<metastore_lsm_probe> 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<metastore_lsm_probe> probe_;
+
     ss::gate gate_;
     ss::abort_source& as_;
     ss::scheduling_group sg_;