storage: include tiered_cloud partitions in disk space management

tiered_cloud partition local logs were excluded from the disk space
manager, so they were never reclaimed under disk pressure. A new
is_cloud_gc_active() predicate (archival retention or tiered_cloud) gates
the space manager's reclaim paths, so tiered_cloud partitions are now
collected below the reconciled L1 offset, like legacy tiered storage.

Also updates disk_log_impl::compute_gc_offset to reset the space management
GC hint on every invocation where it is present.

Signed-off-by: Oren Leiman <oren.leiman@redpanda.com>

Author: oleiman

src/v/resource_mgmt/storage.cc

@@ -289,7 +289,9 @@ eviction_policy::collect_reclaimable_offsets() {
      */
     chunked_vector<ss::lw_shared_ptr<cluster::partition>> partitions;
     for (const auto& p : _pm->local().partitions()) {
-        if (!p.second->remote_partition()) {
+        if (
+          !p.second->remote_partition()
+          && !p.second->log()->config().is_tiered_cloud()) {
             continue;
         }
         partitions.push_back(p.second);

src/v/storage/disk_log_impl.cc

@@ -1315,6 +1315,10 @@ bool disk_log_impl::is_cloud_retention_active() const {
            && (config().is_archival_enabled());
 }
 
+bool disk_log_impl::is_cloud_gc_active() const {
+    return is_cloud_retention_active() || config().is_tiered_cloud();
+}
+
 /*
  * applies overrides for non-cloud storage settings
  */
@@ -1921,7 +1925,7 @@ disk_log_impl::compute_gc_offset(gc_config cfg) {
     // local-target override engages there under retention_local_strict.
     cfg = apply_kafka_retention_overrides(cfg);
     if (_cloud_gc_offset.has_value()) {
-        co_return _cloud_gc_offset;
+        co_return std::exchange(_cloud_gc_offset, std::nullopt);
     }
     co_await maybe_adjust_retention_timestamps();
     cfg = maybe_apply_local_storage_overrides(cfg);
@@ -3921,7 +3925,7 @@ disk_log_impl::disk_usage_and_reclaimable_space(gc_config input_cfg) {
           && seg->offsets().get_dirty_offset() <= retention_offset.value()) {
             retention_segments.push_back(seg);
         } else if (
-          is_cloud_retention_active()
+          is_cloud_gc_active()
           && seg->offsets().get_dirty_offset() <= max_removable) {
             available_segments.push_back(seg);
         } else {
@@ -3937,8 +3941,8 @@ disk_log_impl::disk_usage_and_reclaimable_space(gc_config input_cfg) {
          * get_reclaimable_offsets is going to be merged together.
          */
         if (
-          !config().is_read_replica_mode_enabled()
-          && is_cloud_retention_active() && seg != _segs.back()
+          !config().is_read_replica_mode_enabled() && is_cloud_gc_active()
+          && seg != _segs.back()
           && seg->offsets().get_dirty_offset() <= max_removable
           && local_retention_offset.has_value()
           && seg->offsets().get_dirty_offset()
@@ -4279,8 +4283,8 @@ ss::future<usage_report> disk_log_impl::disk_usage(gc_config cfg) {
 chunked_vector<ss::lw_shared_ptr<segment>>
 disk_log_impl::cloud_gc_eligible_segments() {
     vassert(
-      is_cloud_retention_active(),
-      "Expected {} to have cloud retention enabled",
+      is_cloud_gc_active(),
+      "Expected cloud GC to be active for {}",
       config().ntp());
 
     constexpr size_t keep_segs = 1;
@@ -4313,7 +4317,7 @@ disk_log_impl::cloud_gc_eligible_segments() {
 }
 
 void disk_log_impl::set_cloud_gc_offset(model::offset offset) {
-    if (!is_cloud_retention_active()) {
+    if (!is_cloud_gc_active()) {
         vlog(
           stlog.debug,
           "Ignoring request to set GC offset on non-cloud enabled partition "
@@ -4338,7 +4342,7 @@ disk_log_impl::get_reclaimable_offsets(gc_config cfg) {
 
     reclaimable_offsets res;
 
-    if (!is_cloud_retention_active()) {
+    if (!is_cloud_gc_active()) {
         vlog(
           stlog.debug,
           "Reporting no reclaimable offsets for non-cloud partition {}",
@@ -4540,7 +4544,7 @@ size_t disk_log_impl::reclaimable_size_bytes() const {
      * local retention size may change. catch these before reporting potentially
      * stale information.
      */
-    if (!is_cloud_retention_active()) {
+    if (!is_cloud_gc_active()) {
         return 0;
     }
     if (config().is_read_replica_mode_enabled()) {

src/v/storage/disk_log_impl.h

@@ -210,9 +210,9 @@ class disk_log_impl final : public log {
     /// Applies retention overrides (callers need not pre-apply them) and
     /// adjusts bogus (future) retention timestamps, mutating segment indexes
     /// when an entire segment is bogus, then returns the offset GC would
-    /// evict to -- usually derived from local retention, but when space
-    /// management has set _cloud_gc_offset that offset is returned instead
-    /// (without resetting it -- that is do_gc's responsibility).
+    /// evict to. Usually derived from local retention; when space management
+    /// has set _cloud_gc_offset, that offset is returned and consumed (reset)
+    /// here, so the caller is responsible for acting on it.
     ss::future<std::optional<model::offset>>
     compute_gc_offset(gc_config cfg) final;
 
@@ -422,6 +422,10 @@ class disk_log_impl final : public log {
     gc_config apply_local_storage_overrides(gc_config) const;
 
     bool is_cloud_retention_active() const;
+    // True when local segments are a reclaimable cache of cloud-resident data
+    // (legacy tiered storage or tiered_cloud); broader than
+    // is_cloud_retention_active(), which is archival-only.
+    bool is_cloud_gc_active() const;
 
     // returns retention_offset(cfg) but may also first apply adjustments to
     // future timestamps if this option is turned on in configuration.