diff --git a/src/v/cloud_topics/level_zero/stm/ctp_stm.cc b/src/v/cloud_topics/level_zero/stm/ctp_stm.cc index 71e1ea732f3ef..fea23c7be9477 100644 --- a/src/v/cloud_topics/level_zero/stm/ctp_stm.cc +++ b/src/v/cloud_topics/level_zero/stm/ctp_stm.cc @@ -116,8 +116,9 @@ ss::future<> ctp_stm::prefix_truncate_bg() { // trims aggressively (the local log only holds placeholders); // storage.mode=tiered_cloud honors local retention plus the compaction // floor and so keeps more data locally. + const bool is_tiered = _raft->log()->config().is_tiered_cloud(); model::offset target; - if (_raft->log()->config().is_tiered_cloud()) { + if (is_tiered) { target = co_await compute_local_retention_offset(); } else { // storage.mode=cloud keeps only placeholders locally; the @@ -133,10 +134,12 @@ ss::future<> ctp_stm::prefix_truncate_bg() { _raft->last_snapshot_index()); try { if ( - _raft->last_snapshot_index() >= target - && _active_readers.empty()) { + _raft->last_snapshot_index() >= target && _active_readers.empty() + && !is_tiered) { // Only wait without a timeout if there are no active readers - // that could be holding us back. + // that could be holding us back. tiered_cloud always polls + // because its space-management offset is set without signalling + // _lro_advanced. co_await _lro_advanced.wait(); } else { co_await _lro_advanced.wait(retry_backoff_time); diff --git a/src/v/resource_mgmt/storage.cc b/src/v/resource_mgmt/storage.cc index fbb8cd76d657a..61c6a357b2150 100644 --- a/src/v/resource_mgmt/storage.cc +++ b/src/v/resource_mgmt/storage.cc @@ -289,7 +289,9 @@ eviction_policy::collect_reclaimable_offsets() { */ chunked_vector> 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); diff --git a/src/v/storage/disk_log_impl.cc b/src/v/storage/disk_log_impl.cc index d373f54624198..3d15c3ebbe314 100644 --- a/src/v/storage/disk_log_impl.cc +++ b/src/v/storage/disk_log_impl.cc @@ -1228,9 +1228,9 @@ disk_log_impl::maybe_apply_local_storage_overrides(gc_config cfg) const { // cloud_retention is disabled, do not override. Cloud-topic partitions // (storage.mode in {cloud, tiered_cloud}) bypass this gate: - // is_cloud_retention_active() is false for them, but ctp_stm still needs + // is_archival_active() is false for them, but ctp_stm still needs // the local-target override to engage under retention_local_strict. - if (!is_cloud_retention_active() && !config().cloud_topic_enabled()) { + if (!is_archival_active() && !config().cloud_topic_enabled()) { return cfg; } @@ -1310,11 +1310,15 @@ gc_config disk_log_impl::apply_local_storage_overrides(gc_config cfg) const { return cfg; } -bool disk_log_impl::is_cloud_retention_active() const { +bool disk_log_impl::is_archival_active() const { return config::shard_local_cfg().cloud_storage_enabled() && (config().is_archival_enabled()); } +bool disk_log_impl::is_cloud_gc_active() const { + return is_archival_active() || config().is_tiered_cloud(); +} + /* * applies overrides for non-cloud storage settings */ @@ -1769,6 +1773,10 @@ ss::future<> disk_log_impl::gc(gc_config cfg) { ss::future> disk_log_impl::do_gc(gc_config cfg) { vassert(!_closed, "gc on closed log - {}", *this); + vassert( + !config().cloud_topic_enabled(), + "[{}] gc on cloud topic partition", + config().ntp()); cfg = apply_overrides(cfg); @@ -1783,7 +1791,7 @@ ss::future> disk_log_impl::do_gc(gc_config cfg) { const auto offset = _cloud_gc_offset.value(); _cloud_gc_offset.reset(); - if (!is_cloud_retention_active()) { + if (!is_archival_active()) { vlog( gclog.warn, "[{}] expected remote retention to be active", @@ -1916,12 +1924,12 @@ disk_log_impl::compute_gc_offset(gc_config cfg) { // ctp_stm for cloud-topic partitions. The offset is retention-driven // unless space management has pinned _cloud_gc_offset, which then takes // precedence. maybe_apply_local_storage_overrides - // bypasses the is_cloud_retention_active() gate for cloud-topic partitions + // bypasses the is_archival_active() gate for cloud-topic partitions // (the gate is false for storage.mode in {cloud, tiered_cloud}), so the // 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 +3929,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 +3945,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 +4287,8 @@ ss::future disk_log_impl::disk_usage(gc_config cfg) { chunked_vector> 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 +4321,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 +4346,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 +4548,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()) { diff --git a/src/v/storage/disk_log_impl.h b/src/v/storage/disk_log_impl.h index 7927129a5cb8f..85093de19110a 100644 --- a/src/v/storage/disk_log_impl.h +++ b/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> compute_gc_offset(gc_config cfg) final; @@ -421,7 +421,11 @@ class disk_log_impl final : public log { gc_config maybe_apply_local_storage_overrides(gc_config) const; gc_config apply_local_storage_overrides(gc_config) const; - bool is_cloud_retention_active() const; + bool is_archival_active() const; + // True when local segments are a reclaimable cache of cloud-resident data + // (legacy tiered storage or tiered_cloud); broader than + // is_archival_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. diff --git a/src/v/storage/log_manager.cc b/src/v/storage/log_manager.cc index 13d3e4478017b..07350794e01e2 100644 --- a/src/v/storage/log_manager.cc +++ b/src/v/storage/log_manager.cc @@ -292,6 +292,13 @@ log_manager::housekeeping_scan(model::timestamp collection_threshold) { current_log.flags |= bflags::compaction_checked; + // Cloud topics manage their local log via ctp_stm (compaction is at + // L1), so skip gc/compaction housekeeping for them. segment.ms rolling + // ran above and still applies. + if (current_log.handle->config().cloud_topic_enabled()) { + continue; + } + // Hold the housekeeping gate to prevent issues with concurrent removal // of the log meta. auto gate = current_log.housekeeping_gate.hold(); @@ -510,6 +517,12 @@ ss::future<> log_manager::gc_loop() { continue; } + // Cloud topics are gc'd by ctp_stm, not here; exclude them from + // the reclaim estimate (segment.ms above still applies). + if (log_meta.handle->config().cloud_topic_enabled()) { + continue; + } + auto ntp = log_meta.handle->config().ntp(); auto usage = co_await log_meta.handle->disk_usage( gc_config(lowest_ts_to_retain(), _config.retention_bytes())); diff --git a/tests/rptest/tests/tiered_cloud_local_retention_test.py b/tests/rptest/tests/tiered_cloud_local_retention_test.py index fffa76c5ddc3a..4c499e7b8b6dd 100644 --- a/tests/rptest/tests/tiered_cloud_local_retention_test.py +++ b/tests/rptest/tests/tiered_cloud_local_retention_test.py @@ -299,6 +299,138 @@ def test_local_data_grows_in_tiered_cloud(self): f"expected near {replication * self.local_target_bytes} bytes" ) + @cluster(num_nodes=4) + def test_capacity_pressure_reclaims_tiered_cloud(self): + """ + Cluster-wide disk pressure (the disk_space_manager capacity path) + reclaims a tiered_cloud partition's local log, not just the + retention.local.target path. + + The topic sets no local target and effectively infinite retention, so + only the capacity path can bound it. On a build that excludes + tiered_cloud partitions from disk_space_manager the local log grows + unbounded and this fails; with them admitted, the eviction policy pins + _cloud_gc_offset and ctp_stm trims to it, so the footprint converges + near the capacity target. + """ + assert self.redpanda is not None + + # Node-wide log-data target, small enough that the produced topic must + # be trimmed to meet it. disk_reservation_percent=0 and percent=100 so + # the bytes target is the effective one. + capacity_target = 16 * 1024 * 1024 + self.redpanda.set_cluster_config( + { + "retention_local_target_capacity_bytes": capacity_target, + "retention_local_target_capacity_percent": 100, + "disk_reservation_percent": 0, + } + ) + + # No retention.local.target.bytes: the local-target path stays inert + # (only the 1-day default applies, and nothing is that old), so only + # the capacity path can bound this topic. retention.bytes is + # effectively unlimited, mirroring the field scenario that filled the + # disk. + rpk = RpkTool(self.redpanda) + rpk.create_topic( + topic=self.topic_name, + partitions=1, + replicas=3, + config={ + TopicSpec.PROPERTY_STORAGE_MODE: (TopicSpec.STORAGE_MODE_TIERED_CLOUD), + "cleanup.policy": TopicSpec.CLEANUP_DELETE, + "retention.bytes": str(self.retention_bytes), + "segment.bytes": str(self.segment_size), + }, + ) + self._wait_for_partition_info() + + # Produce well past the capacity target so an untrimmed build grows far + # beyond it. + produce_bytes = 64 * 1024 * 1024 + self._produce(produce_bytes) + self.wait_until_reconciled(topic=self.topic_name, partition=0) + + # The target is per node; the metric sums this topic's 3 replicas and + # excludes internal logs (topic-filtered). Allow 3x the target plus + # per-replica headroom for the never-evicted active segment and + # trim-cycle latency. Still far below the ~3x produce an untrimmed + # build retains. + replication = 3 + ceiling = replication * (capacity_target + 4 * self.segment_size) + # The topic has no local retention target and a lenient retention.bytes, + # so ctp_stm's own retention never trims at this volume; only a + # disk_space_manager capacity pin can drive the reclaim. Staying below + # the ceiling therefore proves the capacity path engaged for a + # tiered_cloud partition. + self._wait_local_below(ceiling_bytes=ceiling, timeout_sec=180) + + @cluster(num_nodes=4) + def test_idle_partition_reclaims_on_pin(self): + """ + A tiered_cloud partition that has gone idle still reclaims when disk + pressure sets _cloud_gc_offset afterwards. + + With no produce the LRO is quiescent, so nothing signals ctp_stm's + truncate loop. It wakes on its bounded poll interval, re-reads the + offset disk_space_manager set, and trims to it, so an idle partition + reclaims without further produce. + """ + assert self.redpanda is not None + + capacity_target = 16 * 1024 * 1024 + replication = 3 + ceiling = replication * (capacity_target + 4 * self.segment_size) + + # Capacity path effectively off: percent=100 with no bytes target lets + # the backlog accumulate without pressure, so nothing trims until the + # bytes target is set below. + self.redpanda.set_cluster_config( + { + "retention_local_target_capacity_percent": 100, + "disk_reservation_percent": 0, + } + ) + + rpk = RpkTool(self.redpanda) + rpk.create_topic( + topic=self.topic_name, + partitions=1, + replicas=3, + config={ + TopicSpec.PROPERTY_STORAGE_MODE: (TopicSpec.STORAGE_MODE_TIERED_CLOUD), + "cleanup.policy": TopicSpec.CLEANUP_DELETE, + "retention.bytes": str(self.retention_bytes), + "segment.bytes": str(self.segment_size), + }, + ) + self._wait_for_partition_info() + + # Fill the local log, then let reconciliation catch up so the LRO is + # quiescent (nothing to drive the truncate loop). Nothing trims yet + # (no pressure, lenient retention.bytes), so the footprint holds near + # the produced size. + self._produce(64 * 1024 * 1024) + self.wait_until_reconciled(topic=self.topic_name, partition=0) + + # Confirm the backlog is present and untrimmed before applying pressure, + # so the reclaim assertion below cannot pass trivially. + wait_until( + lambda: self._local_partition_bytes() > ceiling, + timeout_sec=60, + backoff_sec=2, + err_msg="backlog never reached the expected size before pressure", + ) + + # Apply pressure: disk_space_manager sets _cloud_gc_offset on its next + # cycle. The LRO is quiescent, so the truncate loop only observes it on + # its next bounded poll, then trims to it. + self.redpanda.set_cluster_config( + {"retention_local_target_capacity_bytes": capacity_target} + ) + self._wait_local_below(ceiling_bytes=ceiling, timeout_sec=120) + @cluster(num_nodes=4) def test_compact_topic_evicts_via_l1_floor(self): """