[v24.3.x] storage: add tombstones_removed metric to probe

src/v/storage/probe.cc

@@ -195,6 +195,12 @@ void probe::setup_metrics(const model::ntp& ntp) {
           [this] { return _compaction_removed_bytes; },
           sm::description("Number of bytes removed by a compaction operation"),
           labels),
+        sm::make_counter(
+          "tombstones_removed",
+          [this] { return _tombstones_removed; },
+          sm::description("Number of tombstone records removed by compaction "
+                          "due to the delete.retention.ms setting."),
+          labels),
       },
       {},
       {sm::shard_label, partition_label});

src/v/storage/probe.h

@@ -97,6 +97,8 @@ class probe {
         _cached_batches_read += batches;
     }
 
+    void add_removed_tombstone() { ++_tombstones_removed; }
+
     void batch_parse_error() { ++_batch_parse_errors; }
 
     void setup_metrics(const model::ntp&);
@@ -137,6 +139,7 @@ class probe {
     uint32_t _batch_parse_errors = 0;
     uint32_t _batch_write_errors = 0;
     double _compaction_ratio = 1.0;
+    uint64_t _tombstones_removed = 0;
 
     ssize_t _compaction_removed_bytes = 0;
 

src/v/storage/segment_deduplication_utils.cc

@@ -190,6 +190,7 @@ ss::future<index_state> deduplicate_segment(
     auto copy_reducer = internal::copy_data_segment_reducer(
       [&map,
        &may_have_tombstone_records,
+       &probe,
        segment_last_offset = seg->offsets().get_committed_offset(),
        past_tombstone_delete_horizon,
        compaction_placeholder_enabled](
@@ -213,6 +214,7 @@ ss::future<index_state> deduplicate_segment(
 
           // Deal with tombstone record removal
           if (r.is_tombstone() && past_tombstone_delete_horizon) {
+              probe.add_removed_tombstone();
               return ss::make_ready_future<bool>(false);
           }
 

src/v/storage/segment_utils.cc

@@ -402,26 +402,27 @@ ss::future<storage::index_state> do_copy_segment_data(
       seg->reader().filename(),
       tmpname);
     bool may_have_tombstone_records = false;
-    auto should_keep = [compacted_list = std::move(compacted_offsets),
-                        past_tombstone_delete_horizon,
-                        &may_have_tombstone_records](
-                         const model::record_batch& b,
-                         const model::record& r,
-                         bool) {
-        // Deal with tombstone record removal
-        if (r.is_tombstone() && past_tombstone_delete_horizon) {
-            return ss::make_ready_future<bool>(false);
-        }
+    auto should_keep =
+      [compacted_list = std::move(compacted_offsets),
+       past_tombstone_delete_horizon,
+       &may_have_tombstone_records,
+       &pb](const model::record_batch& b, const model::record& r, bool) {
+          // Deal with tombstone record removal
+          if (r.is_tombstone() && past_tombstone_delete_horizon) {
+              pb.add_removed_tombstone();
+              return ss::make_ready_future<bool>(false);
+          }
 
-        const auto o = b.base_offset() + model::offset_delta(r.offset_delta());
-        const auto keep = compacted_list.contains(o);
+          const auto o = b.base_offset()
+                         + model::offset_delta(r.offset_delta());
+          const auto keep = compacted_list.contains(o);
 
-        if (r.is_tombstone() && keep) {
-            may_have_tombstone_records = true;
-        }
+          if (r.is_tombstone() && keep) {
+              may_have_tombstone_records = true;
+          }
 
-        return ss::make_ready_future<bool>(keep);
-    };
+          return ss::make_ready_future<bool>(keep);
+      };
 
     model::offset segment_last_offset{};
     if (likely(feature_table.local().is_active(