ct/l1: add worker_probe

Used at the `compaction_worker` level (i.e across all shards), as well
as within the `compaction_source` class to capture statistics about
removed data.

Author: WillemKauf

src/v/cloud_topics/level_one/compaction/BUILD

@@ -55,6 +55,7 @@ redpanda_cc_library(
         ":filter",
         ":logger",
         ":meta",
+        ":worker_probe",
         "//src/v/bytes",
         "//src/v/bytes:iostream",
         "//src/v/cloud_topics:log_reader_config",
@@ -95,6 +96,7 @@ redpanda_cc_library(
         ":meta",
         ":scheduler_probe",
         ":source_and_sink",
+        ":worker_probe",
         "//src/v/cloud_topics/level_one/common:abstract_io",
         "//src/v/cloud_topics/level_one/common:file_io",
         "//src/v/cloud_topics/level_one/metastore",
@@ -289,3 +291,20 @@ redpanda_cc_library(
         "@seastar",
     ],
 )
+
+redpanda_cc_library(
+    name = "worker_probe",
+    srcs = [
+        "worker_probe.cc",
+    ],
+    hdrs = [
+        "worker_probe.h",
+    ],
+    deps = [
+        "//src/v/compaction:types",
+        "//src/v/config",
+        "//src/v/metrics",
+        "//src/v/utils:log_hist",
+        "@seastar",
+    ],
+)

src/v/cloud_topics/level_one/compaction/source.cc

@@ -138,7 +138,8 @@ compaction_source::compaction_source(
   metastore* metastore,
   io* io,
   ss::abort_source& as,
-  compaction_job_state& state)
+  compaction_job_state& state,
+  compaction_worker_probe& probe)
   : _ntp(std::move(ntp))
   , _tp(tp)
   , _dirty_range_intervals(dirty_range_intervals)
@@ -158,7 +159,8 @@ compaction_source::compaction_source(
   , _metastore(metastore)
   , _io(io)
   , _as(as)
-  , _state(state) {}
+  , _state(state)
+  , _probe(probe) {}
 
 ss::future<> compaction_source::initialize() { co_return; }
 
@@ -295,6 +297,8 @@ ss::future<ss::stop_iteration> compaction_source::deduplication_iteration(
               last_offset,
               stats);
         }
+
+        _probe.add_stats(stats);
     }
 
     co_return ss::stop_iteration::no;

src/v/cloud_topics/level_one/compaction/source.h

@@ -13,6 +13,7 @@
 #include "cloud_topics/level_one/common/abstract_io.h"
 #include "cloud_topics/level_one/compaction/filter.h"
 #include "cloud_topics/level_one/compaction/meta.h"
+#include "cloud_topics/level_one/compaction/worker_probe.h"
 #include "cloud_topics/level_one/metastore/extent_metadata_reader.h"
 #include "cloud_topics/level_one/metastore/metastore.h"
 #include "cloud_topics/level_one/metastore/offset_interval_set.h"
@@ -36,7 +37,8 @@ class compaction_source : public compaction::sliding_window_reducer::source {
       metastore*,
       io*,
       ss::abort_source&,
-      compaction_job_state&);
+      compaction_job_state&,
+      compaction_worker_probe&);
     ss::future<> initialize() final;
     ss::future<ss::stop_iteration> map_building_iteration() final;
     ss::future<ss::stop_iteration>
@@ -92,6 +94,7 @@ class compaction_source : public compaction::sliding_window_reducer::source {
 
     ss::abort_source& _as;
     compaction_job_state& _state;
+    compaction_worker_probe& _probe;
 
     // Dirty ranges returned by the `metastore` that were indexed during
     // `map_deduplication_iteration`.

src/v/cloud_topics/level_one/compaction/tests/BUILD

@@ -39,6 +39,7 @@ redpanda_cc_gtest(
         "//src/v/cloud_topics/level_one/compaction:filter",
         "//src/v/cloud_topics/level_one/compaction:meta",
         "//src/v/cloud_topics/level_one/compaction:source_and_sink",
+        "//src/v/cloud_topics/level_one/compaction:worker_probe",
         "//src/v/cloud_topics/level_one/frontend_reader/tests:l1_reader_fixture",
         "//src/v/cloud_topics/level_one/metastore",
         "//src/v/cloud_topics/level_one/metastore:simple",

src/v/cloud_topics/level_one/compaction/tests/reducer_test.cc

@@ -20,6 +20,7 @@
 #include "cloud_topics/level_one/compaction/sink.h"
 #include "cloud_topics/level_one/compaction/source.h"
 #include "cloud_topics/level_one/compaction/tests/in_memory_sink.h"
+#include "cloud_topics/level_one/compaction/worker_probe.h"
 #include "cloud_topics/level_one/frontend_reader/tests/l1_reader_fixture.h"
 #include "cloud_topics/level_one/metastore/metastore.h"
 #include "cloud_topics/level_one/metastore/simple_metastore.h"
@@ -104,6 +105,7 @@ ss::future<> do_compact(
     auto state = l1::compaction_job_state::running;
     auto map = compaction::simple_key_offset_map();
     auto dirty_range_intervals = offsets_response.dirty_ranges.to_vec();
+    l1::compaction_worker_probe probe;
     auto src = std::make_unique<l1::compaction_source>(
       ntp,
       tidp,
@@ -115,7 +117,8 @@ ss::future<> do_compact(
       metastore,
       io,
       as,
-      state);
+      state,
+      probe);
     auto sink = std::make_unique<l1::compaction_sink>(
       tidp,
       dirty_range_intervals,

src/v/cloud_topics/level_one/compaction/worker.cc

@@ -45,6 +45,7 @@ compaction_worker::compaction_worker(
   , _metadata_cache(metadata_cache) {}
 
 ss::future<> compaction_worker::start() {
+    _probe.setup_metrics();
     start_work_loop();
     co_return;
 }
@@ -207,7 +208,8 @@ ss::future<> compaction_worker::compact_log(log_compaction_meta* log) {
       _metastore,
       _io,
       _as,
-      _job_state);
+      _job_state,
+      _probe);
     auto sink = std::make_unique<compaction_sink>(
       tidp,
       dirty_range_intervals,
@@ -219,6 +221,9 @@ ss::future<> compaction_worker::compact_log(log_compaction_meta* log) {
     auto reducer = compaction::sliding_window_reducer(
       std::move(src), std::move(sink));
 
+    // Start measuring time-to-compact here.
+    auto m = _probe.auto_compaction_measurement();
+
     auto compact_fut = co_await ss::coroutine::as_future(
       std::move(reducer).run());
 
@@ -232,6 +237,9 @@ ss::future<> compaction_worker::compact_log(log_compaction_meta* log) {
           "Caught exception {} while compacting CTP {}.",
           eptr,
           tidp);
+
+        // Don't let failed compaction runs contribute to the histogram.
+        m->cancel();
     } else {
         vlog(compaction_log.info, "Finished compacting CTP {}", tidp);
     }

src/v/cloud_topics/level_one/compaction/worker.h

@@ -14,6 +14,7 @@
 #include "cloud_topics/level_one/compaction/committer.h"
 #include "cloud_topics/level_one/compaction/meta.h"
 #include "cloud_topics/level_one/compaction/source.h"
+#include "cloud_topics/level_one/compaction/worker_probe.h"
 #include "cloud_topics/level_one/metastore/metastore.h"
 #include "cluster/metadata_cache.h"
 #include "compaction/key_offset_map.h"
@@ -189,6 +190,8 @@ class compaction_worker {
     compaction_committer* _committer;
 
     cluster::metadata_cache* _metadata_cache;
+
+    compaction_worker_probe _probe;
 };
 
 } // namespace cloud_topics::l1

src/v/cloud_topics/level_one/compaction/worker_probe.cc

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2025 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/compaction/worker_probe.h"
+
+#include "config/configuration.h"
+#include "metrics/metrics.h"
+#include "metrics/prometheus_sanitize.h"
+
+#include <seastar/core/metrics.hh>
+
+namespace cloud_topics::l1 {
+
+void compaction_worker_probe::setup_metrics() {
+    namespace sm = ss::metrics;
+
+    if (config::shard_local_cfg().disable_metrics()) {
+        return;
+    }
+
+    _metrics.add_group(
+      prometheus_sanitize::metrics_name("cloud_topics:compaction_worker"),
+      {
+        sm::make_gauge(
+          "batches_removed",
+          [this] { return _batches_removed; },
+          sm::description(
+            "Number of batches removed across all cloud topic partitions on "
+            "this shard")),
+        sm::make_gauge(
+          "records_removed",
+          [this] { return _records_removed; },
+          sm::description(
+            "Number of records removed across all cloud topic partitions on "
+            "this shard")),
+        sm::make_gauge(
+          "tombstones_removed",
+          [this] { return _tombstones_removed; },
+          sm::description(
+            "Number of tombstone records removed across all cloud topic "
+            "partitions on this shard")),
+        sm::make_histogram(
+          "compaction_duration_seconds",
+          [this] { return _compaction_runs.internal_histogram_logform(); },
+          sm::description(
+            "The duration of a compaction run for cloud topic partitions on "
+            "this shard")),
+      });
+}
+
+} // namespace cloud_topics::l1

src/v/cloud_topics/level_one/compaction/worker_probe.h

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2025 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 "compaction/types.h"
+#include "metrics/metrics.h"
+#include "utils/log_hist.h"
+
+#include <seastar/core/metrics_registration.hh>
+
+#include <cstdint>
+#include <memory>
+
+namespace cloud_topics::l1 {
+
+class compaction_worker_probe {
+public:
+    using hist_t = log_hist_internal;
+    compaction_worker_probe() = default;
+
+    compaction_worker_probe(const compaction_worker_probe&) = delete;
+    compaction_worker_probe& operator=(const compaction_worker_probe&) = delete;
+    compaction_worker_probe(compaction_worker_probe&&) = delete;
+    compaction_worker_probe& operator=(compaction_worker_probe&&) = delete;
+    ~compaction_worker_probe() = default;
+
+    void setup_metrics();
+
+    std::unique_ptr<hist_t::measurement> auto_compaction_measurement() {
+        return _compaction_runs.auto_measure();
+    }
+
+    void add_stats(const compaction::stats& stats) {
+        _batches_removed += stats.batches_discarded;
+        _records_removed += stats.records_discarded;
+        _tombstones_removed += stats.expired_tombstones_discarded;
+    }
+
+private:
+    hist_t _compaction_runs;
+
+    uint64_t _batches_removed{0};
+    uint64_t _records_removed{0};
+    uint64_t _tombstones_removed{0};
+
+    metrics::internal_metric_groups _metrics;
+};
+
+} // namespace cloud_topics::l1