admin: expose time point for last completed scrub

When a full scrub of the log finishes, via one or more scrubs, persist
the timestamp at which this happened. This timestamp is then exposed via
the admin API. We will use this in our ducktape tests for end of test
scrubbing.

src/v/cloud_storage/partition_manifest.cc

@@ -2732,6 +2732,10 @@ void partition_manifest::process_anomalies(
     _last_partition_scrub = scrub_timestamp;
     _last_scrubbed_offset = last_scrubbed_offset;
 
+    if (!_last_scrubbed_offset) {
+        _detected_anomalies.last_complete_scrub = scrub_timestamp;
+    }
+
     vlog(
       cst_log.debug,
       "[{}] Anomalies processed: {{ detected: {}, last_partition_scrub: {}, "

src/v/cloud_storage/tests/anomalies_detector_test.cc

@@ -23,6 +23,19 @@
 #include <boost/test/tools/old/interface.hpp>
 #include <boost/test/unit_test.hpp>
 
+namespace cloud_storage {
+
+bool operator==(const anomalies& lhs, const anomalies& rhs) {
+    return lhs.missing_partition_manifest == rhs.missing_partition_manifest
+           && lhs.missing_spillover_manifests == rhs.missing_spillover_manifests
+           && lhs.missing_segments == rhs.missing_segments
+           && lhs.segment_metadata_anomalies == rhs.segment_metadata_anomalies;
+
+    // anomalies::last_complete_scrub is intentionally omitted
+}
+
+} // namespace cloud_storage
+
 namespace {
 
 ss::logger test_logger{"anomaly_detection_test"};

src/v/cloud_storage/types.cc

@@ -264,6 +264,9 @@ anomalies& anomalies::operator+=(anomalies&& other) {
       std::make_move_iterator(other.segment_metadata_anomalies.begin()),
       std::make_move_iterator(other.segment_metadata_anomalies.end()));
 
+    last_complete_scrub = std::max(
+      last_complete_scrub, other.last_complete_scrub);
+
     return *this;
 }
 

src/v/cloud_storage/types.h

@@ -391,20 +391,24 @@ struct anomalies
     absl::node_hash_set<segment_meta> missing_segments;
     // Segments that have metadata anomalies (e.g. gaps or overlaps)
     segment_meta_anomalies segment_metadata_anomalies;
+    // Optional timestamp indicating the last time point at which
+    // the scrub of the full log completed.
+    std::optional<model::timestamp> last_complete_scrub;
 
     auto serde_fields() {
         return std::tie(
           missing_partition_manifest,
           missing_spillover_manifests,
           missing_segments,
-          segment_metadata_anomalies);
+          segment_metadata_anomalies,
+          last_complete_scrub);
     }
 
     bool has_value() const;
 
     anomalies& operator+=(anomalies&&);
 
-    bool operator==(anomalies const&) const = default;
+    friend bool operator==(const anomalies& lhs, const anomalies& rhs);
 };
 
 std::ostream& operator<<(std::ostream& o, const anomalies& a);

src/v/redpanda/admin/api-doc/shadow_indexing.json

@@ -518,6 +518,10 @@
           "type": "array",
           "items": {"type": "metadata_anomaly"},
           "nullable": true
+        },
+        "last_complete_scrub_at": {
+          "type": "long",
+          "nullable": true
         }
       }
     }

src/v/redpanda/admin_server.cc

@@ -5304,6 +5304,10 @@ map_anomalies_to_json(
     json.partition = ntp.tp.partition();
     json.revision_id = initial_rev();
 
+    if (detected.last_complete_scrub) {
+        json.last_complete_scrub_at = detected.last_complete_scrub->value();
+    }
+
     if (detected.missing_partition_manifest) {
         json.missing_partition_manifest = true;
     }

tests/rptest/tests/cloud_storage_scrubber_test.py

@@ -118,6 +118,8 @@ def _collect_anomalies(self):
                                                           topic=self.topic,
                                                           partition=pid)
 
+            anomalies.pop("last_complete_scrub_at", None)
+
             ntpr = NTPR(ns=anomalies["ns"],
                         topic=anomalies["topic"],
                         partition=anomalies["partition"],