From 88cb2a986ef0776a1eb84b0df9581b3e7ffb189d Mon Sep 17 00:00:00 2001 From: Dragon Roar <2181719471@qq.com> Date: Mon, 26 Oct 2026 07:13:41 +0000 Subject: [PATCH] fix: reclaim HDFS temporary files left behind by a datanode crash (#9616) * fix: reclaim HDFS temporary files left behind by a datanode crash Signed-off-by: lczllx <2181719471@qq.com> * fix(mito2): stop the temporary file sweep from starving GC jobs Signed-off-by: lczllx <2181719471@qq.com> * feat(mito2): add metrics for the temporary file sweep Signed-off-by: lczllx <2181719471@qq.com> * refactor(object-store): share one definition of the temporary file name Signed-off-by: lczllx <2181719471@qq.com> * fix(mito2): skip the temporary file sweep when no GC job is allowed Signed-off-by: lczllx <2181719471@qq.com> * refactor: gate the temporary file sweep behind the hdfs-object-store feature Signed-off-by: lczllx <2181719471@qq.com> * fix(mito2): sweep temporary files when a create opens an existing region Signed-off-by: lczllx <2181719471@qq.com> * docs(config): document the temporary file sweep in the GC settings Signed-off-by: lczllx <2181719471@qq.com> * refactor: align the temporary file sweep code with the surrounding style Signed-off-by: lczllx <2181719471@qq.com> --------- Signed-off-by: lczllx <2181719471@qq.com> --- config/config.md | 2 +- config/datanode.example.toml | 5 +- src/mito2/AGENTS.md | 1 + src/mito2/src/gc.rs | 15 +- src/mito2/src/lib.rs | 2 + src/mito2/src/metrics.rs | 21 ++ src/mito2/src/sst/location.rs | 7 +- src/mito2/src/temp_file_sweeper.rs | 152 ++++++++++ src/mito2/src/worker.rs | 45 +++ src/mito2/src/worker/handle_catchup.rs | 21 +- src/mito2/src/worker/handle_create.rs | 12 + src/object-store/src/factory.rs | 2 + src/object-store/src/layers/hdfs.rs | 48 +++- src/object-store/src/lib.rs | 2 + src/object-store/src/temp_file.rs | 373 +++++++++++++++++++++++++ 15 files changed, 686 insertions(+), 22 deletions(-) create mode 100644 src/mito2/src/temp_file_sweeper.rs create mode 100644 src/object-store/src/temp_file.rs diff --git a/config/config.md b/config/config.md index e370bb157a0..36c6f02baff 100644 --- a/config/config.md +++ b/config/config.md @@ -724,7 +724,7 @@ | `region_engine.mito.gc` | -- | -- | -- | | `region_engine.mito.gc.enable` | Bool | `false` | Whether GC is enabled. Need to be the same with metasrv's `gc.enable` or unexpected behavior will occur | | `region_engine.mito.gc.lingering_time` | String | `1h` | Lingering time before deleting files.
Should be long enough to allow long running queries to finish.
If set to None, then unused files will be deleted immediately. | -| `region_engine.mito.gc.unknown_file_lingering_time` | String | `1d` | Lingering time before deleting unknown files (files with undetermined expel time).
Only applies during full file listing GC.
This uses the object's last-modified timestamp as a heuristic (strict less-than comparison);
do not configure this value too small in production to avoid deleting pre-manifest files
from in-progress compaction or flush.
For active/open regions, an unknown file is deleted only if its object last-modified time exceeds this TTL.
If the object store does not provide a last-modified timestamp, the file is conservatively kept.
For dropped regions, unknown files are deleted immediately. | +| `region_engine.mito.gc.unknown_file_lingering_time` | String | `1d` | Lingering time before deleting unknown files (files with undetermined expel time).
It is also the age threshold of the sweep that reclaims temporary files left behind by
a crashed HDFS writer, and that sweep runs even if `enable` is false. HDFS advances the
modification time of a file under construction only as its blocks complete, so lowering
this value also shrinks the margin between the sweep and files that are still being written.
This uses the object's last-modified timestamp as a heuristic (strict less-than comparison);
do not configure this value too small in production to avoid deleting pre-manifest files
from in-progress compaction or flush.
For active/open regions, an unknown file is deleted only if its object last-modified time exceeds this TTL.
If the object store does not provide a last-modified timestamp, the file is conservatively kept.
For dropped regions, unknown files are deleted immediately. | | `region_engine.file` | -- | -- | Enable the file engine. | | `region_engine.metric` | -- | -- | Metric engine options. | | `logging` | -- | -- | The logging options. | diff --git a/config/datanode.example.toml b/config/datanode.example.toml index bfa3afc0bcf..f6e9a2571f8 100644 --- a/config/datanode.example.toml +++ b/config/datanode.example.toml @@ -831,7 +831,10 @@ enable = false lingering_time = "1h" ## Lingering time before deleting unknown files (files with undetermined expel time). -## Only applies during full file listing GC. +## It is also the age threshold of the sweep that reclaims temporary files left behind by +## a crashed HDFS writer, and that sweep runs even if `enable` is false. HDFS advances the +## modification time of a file under construction only as its blocks complete, so lowering +## this value also shrinks the margin between the sweep and files that are still being written. ## This uses the object's last-modified timestamp as a heuristic (strict less-than comparison); ## do not configure this value too small in production to avoid deleting pre-manifest files ## from in-progress compaction or flush. diff --git a/src/mito2/AGENTS.md b/src/mito2/AGENTS.md index 811632b1e26..a02fd4dfd43 100644 --- a/src/mito2/AGENTS.md +++ b/src/mito2/AGENTS.md @@ -34,6 +34,7 @@ snapshot isolation). It implements the `RegionEngine` trait from `store-api`. | `schedule` | `src/mito2/src/schedule/` | Local/remote background job scheduling | | `remap_manifest` | `src/mito2/src/remap_manifest.rs` | Manifest path remapping for region copy/migration | | `config` | `src/mito2/src/config.rs` | `MitoConfig` tuning knobs | +| `temp_file_sweeper` | `src/mito2/src/temp_file_sweeper.rs` | Reclaims temporary files a crashed writer left in a region, triggered when the region becomes a writable leader or a create opens an existing region (under the `hdfs-object-store` feature) | | `test_util` | `src/mito2/src/test_util.rs` | `TestEnv` and builders (under the `test` feature) | ## Write path diff --git a/src/mito2/src/gc.rs b/src/mito2/src/gc.rs index a5474099cbf..541bb4872b4 100644 --- a/src/mito2/src/gc.rs +++ b/src/mito2/src/gc.rs @@ -159,6 +159,12 @@ pub struct GcConfig { /// expel time is the time when the file is considered as removed, as in removed from the manifest. /// This should only occur rarely, as manifest keep tracks in `removed_files` field /// unless something goes wrong. + /// + /// It is also the age threshold of the sweep that reclaims temporary files left behind + /// by a crashed HDFS writer, and that sweep runs even if `enable` is false. HDFS advances + /// the modification time of a file under construction only as its blocks complete, so + /// lowering this value also shrinks the margin between the sweep and files that are + /// still being written. #[serde(with = "humantime_serde")] pub unknown_file_lingering_time: Duration, /// Maximum concurrent list operations per GC job. @@ -167,6 +173,8 @@ pub struct GcConfig { /// Maximum concurrent GC jobs. /// This is used to limit the number of concurrent GC jobs running on the datanode /// to prevent too many concurrent GC jobs from overwhelming the datanode. + /// Setting it to 0 disables GC jobs and the temporary file sweep, which is limited + /// to the same number of concurrent tasks. pub max_concurrent_gc_job: usize, } @@ -177,8 +185,9 @@ impl Default for GcConfig { // expect long running queries to be finished(or at least be able to notify it's using a deleted file) within a reasonable time lingering_time: Some(Duration::from_secs(60 * 60)), // 1 day, for unknown expel time, which is when this file get removed from manifest. - // Only applies to full-listing GC for active/open regions. A long default avoids - // accidentally deleting pre-manifest files (e.g. compaction/flush still in progress). + // A long default avoids accidentally deleting pre-manifest files (e.g. + // compaction/flush still in progress) or temporary files of a writer that is + // still alive; the temporary file sweep uses this value as well. unknown_file_lingering_time: Duration::from_secs(24 * 60 * 60), max_concurrent_lister_per_gc_job: 32, max_concurrent_gc_job: 4, @@ -806,7 +815,7 @@ impl LocalGcWorker { /// Only `.puffin` files (not subdirectories) are included. async fn list_region_index_files(&self, region_id: RegionId) -> Result> { let region_dir = self.access_layer.build_region_dir(region_id); - let index_dir = object_store::util::join_dir(®ion_dir, "index"); + let index_dir = location::index_dir(®ion_dir); let mut lister = match self .access_layer diff --git a/src/mito2/src/lib.rs b/src/mito2/src/lib.rs index dcbfe60dea5..a01bca47dd7 100644 --- a/src/mito2/src/lib.rs +++ b/src/mito2/src/lib.rs @@ -42,6 +42,8 @@ pub mod request; pub mod schedule; pub mod series_index; pub mod sst; +#[cfg(feature = "hdfs-object-store")] +mod temp_file_sweeper; mod time_provider; pub mod wal; pub mod worker; diff --git a/src/mito2/src/metrics.rs b/src/mito2/src/metrics.rs index cffa6de39a1..e8718f0e758 100644 --- a/src/mito2/src/metrics.rs +++ b/src/mito2/src/metrics.rs @@ -607,6 +607,27 @@ lazy_static! { register_int_counter!("greptime_mito_flush_file_total", "mito flushed file count").unwrap(); } +#[cfg(feature = "hdfs-object-store")] +lazy_static! { + /// Counter for the number of temporary files deleted by the sweeper. + pub static ref SWEEP_FILES_DELETED_TOTAL: IntCounter = register_int_counter!( + "greptime_mito_sweep_files_deleted_total", + "Total temporary files deleted by the sweeper", + ).unwrap(); + + /// Counter for the number of failures while sweeping temporary files. + pub static ref SWEEP_ERRORS_TOTAL: IntCounter = register_int_counter!( + "greptime_mito_sweep_errors_total", + "Total failures while sweeping temporary files", + ).unwrap(); + + /// Counter for the number of temporary file sweeps run. + pub static ref SWEEP_RUNS_TOTAL: IntCounter = register_int_counter!( + "greptime_mito_sweep_runs_total", + "Total temporary file sweeps run", + ).unwrap(); +} + /// Stager notifier to collect metrics. pub struct StagerMetrics { cache_hit: IntCounter, diff --git a/src/mito2/src/sst/location.rs b/src/mito2/src/sst/location.rs index 5bff4a6d4cf..379d3f3514e 100644 --- a/src/mito2/src/sst/location.rs +++ b/src/mito2/src/sst/location.rs @@ -39,6 +39,11 @@ pub fn region_dir_from_table_dir( } } +/// Generate index dir from region_dir +pub fn index_dir(region_dir: &str) -> String { + util::join_dir(region_dir, "index") +} + pub fn sst_file_path(table_dir: &str, region_file_id: RegionFileId, path_type: PathType) -> String { let region_dir = region_dir_from_table_dir(table_dir, region_file_id.region_id(), path_type); util::join_path( @@ -49,7 +54,7 @@ pub fn sst_file_path(table_dir: &str, region_file_id: RegionFileId, path_type: P pub fn index_file_path(table_dir: &str, index_id: RegionIndexId, path_type: PathType) -> String { let region_dir = region_dir_from_table_dir(table_dir, index_id.file_id.region_id(), path_type); - let index_dir = util::join_dir(®ion_dir, "index"); + let index_dir = index_dir(®ion_dir); let filename = if index_id.version == 0 { format!("{}.puffin", index_id.file_id.file_id()) diff --git a/src/mito2/src/temp_file_sweeper.rs b/src/mito2/src/temp_file_sweeper.rs new file mode 100644 index 00000000000..8cece62aef2 --- /dev/null +++ b/src/mito2/src/temp_file_sweeper.rs @@ -0,0 +1,152 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Reclaims the temporary files that a crashed writer left in a region. + +use std::sync::Arc; +use std::time::Duration; + +use common_telemetry::info; +use object_store::temp_file::{SweepReport, sweep_temporary_files, writes_temporary_files}; +use store_api::storage::RegionId; +use tokio::sync::Semaphore; + +use crate::access_layer::{AccessLayer, AccessLayerRef}; +use crate::manifest::storage::manifest_dir; +use crate::manifest::storage::staging::{staging_blob_path, staging_manifest_path}; +use crate::metrics::{SWEEP_ERRORS_TOTAL, SWEEP_FILES_DELETED_TOTAL, SWEEP_RUNS_TOTAL}; +use crate::sst::location::index_dir; + +/// Returns the directories that `region_id` writes non-append files into. +/// +/// Every one of them can hold the temporary file of a crashed writer, and a +/// list is not recursive, so the sweep has to name each directory. +pub(crate) fn region_write_dirs(access_layer: &AccessLayer, region_id: RegionId) -> Vec { + let region_dir = access_layer.build_region_dir(region_id); + let manifest_path = manifest_dir(®ion_dir); + vec![ + region_dir.clone(), + index_dir(®ion_dir), + manifest_path.clone(), + staging_blob_path(&manifest_path), + staging_manifest_path(&manifest_path), + ] +} + +/// Sweeps the temporary files of `region_id` in the background. +/// +/// The caller is on the path that makes the region its writable leader, so +/// reclaiming leftovers must not slow it down or fail it: everything happens in +/// a spawned task that only logs. +/// +/// `older_than` is the margin the caller takes from GC's +/// `unknown_file_lingering_time`, which guards the same thing: a file this old +/// is not one a writer is still filling. +/// +/// A sweep waits for a permit of `sweep_limiter`, so a `None` means the +/// datanode allows no concurrent GC job and there is no sweep. +pub(crate) fn spawn_region_sweep( + access_layer: AccessLayerRef, + region_id: RegionId, + older_than: Duration, + sweep_limiter: Option>, +) { + // No write to this store leaves a temporary file behind, so don't spend a + // task on it: a sweep would still list every directory of the region to + // find that out. + if !writes_temporary_files(access_layer.object_store()) { + return; + } + + let Some(sweep_limiter) = sweep_limiter else { + return; + }; + + common_runtime::spawn_global(async move { + // A burst of regions becoming writable must not run more sweeps at once + // than the datanode allows GC jobs. Wait for a permit instead of giving + // up, the sweep has no deadline. + let Ok(_permit) = sweep_limiter.acquire_owned().await else { + return; + }; + + let dirs = region_write_dirs(&access_layer, region_id); + let report = sweep_temporary_files(access_layer.object_store(), &dirs, older_than).await; + record_sweep(&report); + if report.deleted > 0 { + info!( + "Reclaimed {} temporary file(s) of a crashed writer in region {region_id}", + report.deleted + ); + } + }); +} + +fn record_sweep(report: &SweepReport) { + SWEEP_RUNS_TOTAL.inc(); + SWEEP_FILES_DELETED_TOTAL.inc_by(report.deleted as u64); + SWEEP_ERRORS_TOTAL.inc_by(report.failed as u64); +} + +#[cfg(test)] +mod tests { + use object_store::ObjectStore; + use object_store::services::Memory; + use store_api::region_request::PathType; + + use super::*; + use crate::test_util::TestEnv; + + fn test_access_layer(env: &TestEnv) -> AccessLayer { + AccessLayer::new( + "test", + PathType::Bare, + ObjectStore::new(Memory::default()).unwrap(), + env.get_puffin_manager(), + env.get_intermediate_manager(), + ) + } + + #[tokio::test] + async fn test_region_write_dirs() { + let env = TestEnv::new().await; + let access_layer = test_access_layer(&env); + let region_id = RegionId::new(1, 2); + + let dirs = region_write_dirs(&access_layer, region_id); + + assert_eq!(5, dirs.len()); + assert!(dirs[0].ends_with("1_0000000002/")); + assert!(dirs[1].ends_with("1_0000000002/index/")); + assert!(dirs[2].ends_with("1_0000000002/manifest/")); + assert!(dirs[3].ends_with("1_0000000002/staging/blob/")); + assert!(dirs[4].ends_with("1_0000000002/staging/manifest/")); + } + + #[test] + fn test_record_sweep() { + let runs = SWEEP_RUNS_TOTAL.get(); + let deleted = SWEEP_FILES_DELETED_TOTAL.get(); + let errors = SWEEP_ERRORS_TOTAL.get(); + + record_sweep(&SweepReport { + deleted: 2, + failed: 1, + }); + + assert_eq!(runs + 1, SWEEP_RUNS_TOTAL.get()); + assert_eq!(deleted + 2, SWEEP_FILES_DELETED_TOTAL.get()); + assert_eq!(errors + 1, SWEEP_ERRORS_TOTAL.get()); + } +} diff --git a/src/mito2/src/worker.rs b/src/mito2/src/worker.rs index c92070ac4e7..789b0f539b1 100644 --- a/src/mito2/src/worker.rs +++ b/src/mito2/src/worker.rs @@ -87,6 +87,8 @@ use crate::sst::file_ref::FileReferenceManagerRef; use crate::sst::index::IndexBuildScheduler; use crate::sst::index::intermediate::IntermediateManager; use crate::sst::index::puffin_manager::PuffinManagerFactory; +#[cfg(feature = "hdfs-object-store")] +use crate::temp_file_sweeper; use crate::time_provider::{StdTimeProvider, TimeProviderRef}; use crate::wal::Wal; use crate::worker::handle_manifest::RegionEditQueues; @@ -241,6 +243,13 @@ impl WorkerGroup { let compaction_memory_manager = Arc::new(new_compaction_memory_manager(compaction_limit_bytes)); let gc_limiter = Arc::new(GcLimiter::new(config.gc.max_concurrent_gc_job)); + // A sweep waits for a permit while a GC job takes one without waiting, so + // sharing the GC limiter would fail GC jobs and report sweeps as GC jobs. + // A datanode that allows no concurrent GC job runs no sweep either: a + // limiter of zero would never hand out the permit the task waits for. + #[cfg(feature = "hdfs-object-store")] + let sweep_limiter = (config.gc.max_concurrent_gc_job > 0) + .then(|| Arc::new(Semaphore::new(config.gc.max_concurrent_gc_job))); let workers = (0..config.num_workers) .map(|id| { @@ -269,6 +278,8 @@ impl WorkerGroup { file_ref_manager: file_ref_manager.clone(), partition_expr_fetcher: partition_expr_fetcher.clone(), flush_semaphore: flush_semaphore.clone(), + #[cfg(feature = "hdfs-object-store")] + sweep_limiter: sweep_limiter.clone(), } .start() }) @@ -491,6 +502,13 @@ impl WorkerGroup { let compaction_memory_manager = Arc::new(new_compaction_memory_manager(compaction_limit_bytes)); let gc_limiter = Arc::new(GcLimiter::new(config.gc.max_concurrent_gc_job)); + // A sweep waits for a permit while a GC job takes one without waiting, so + // sharing the GC limiter would fail GC jobs and report sweeps as GC jobs. + // A datanode that allows no concurrent GC job runs no sweep either: a + // limiter of zero would never hand out the permit the task waits for. + #[cfg(feature = "hdfs-object-store")] + let sweep_limiter = (config.gc.max_concurrent_gc_job > 0) + .then(|| Arc::new(Semaphore::new(config.gc.max_concurrent_gc_job))); let workers = (0..config.num_workers) .map(|id| { WorkerStarter { @@ -518,6 +536,8 @@ impl WorkerGroup { file_ref_manager: file_ref_manager.clone(), partition_expr_fetcher: partition_expr_fetcher.clone(), flush_semaphore: flush_semaphore.clone(), + #[cfg(feature = "hdfs-object-store")] + sweep_limiter: sweep_limiter.clone(), } .start() }) @@ -628,6 +648,10 @@ struct WorkerStarter { file_ref_manager: FileReferenceManagerRef, partition_expr_fetcher: PartitionExprFetcherRef, flush_semaphore: Arc, + #[cfg(feature = "hdfs-object-store")] + /// Limiter for the temporary file sweeps, or `None` if the datanode allows + /// no concurrent GC job. + sweep_limiter: Option>, } impl WorkerStarter { @@ -728,6 +752,8 @@ impl WorkerStarter { partition_expr_fetcher: self.partition_expr_fetcher, flush_semaphore: self.flush_semaphore, plugins: self.plugins, + #[cfg(feature = "hdfs-object-store")] + sweep_limiter: self.sweep_limiter, }; let handle = common_runtime::spawn_global(async move { worker_thread.run().await; @@ -1055,6 +1081,10 @@ struct RegionWorkerLoop { flush_semaphore: Arc, /// Plugins for flush hooks. plugins: Plugins, + #[cfg(feature = "hdfs-object-store")] + /// Limiter for the temporary file sweeps, or `None` if the datanode allows + /// no concurrent GC job. + sweep_limiter: Option>, } impl RegionWorkerLoop { @@ -1475,6 +1505,10 @@ impl RegionWorkerLoop { // changes a writable or staging leader. region.set_role(RegionRole::DowngradingLeader); } + #[cfg(feature = "hdfs-object-store")] + let sweep_older_than = self.config.gc.unknown_file_lingering_time; + #[cfg(feature = "hdfs-object-store")] + let sweep_limiter = self.sweep_limiter.clone(); // We need to do this in background as we need the manifest lock. common_runtime::spawn_global(async move { match region.set_role_state_gracefully(region_role_state).await { @@ -1483,6 +1517,17 @@ impl RegionWorkerLoop { let _ = sender.send(SetRegionRoleStateResponse::success( SetRegionRoleStateSuccess::mito(last_entry_id), )); + // Only a writable leader reclaims what a writer that died + // before it renamed its temporary file left behind. + #[cfg(feature = "hdfs-object-store")] + if region_role_state == SettableRegionRoleState::Leader { + temp_file_sweeper::spawn_region_sweep( + region.access_layer(), + region_id, + sweep_older_than, + sweep_limiter, + ); + } } Err(e) => { error!(e; "Failed to set region {} role state to {:?}", region_id, region_role_state); diff --git a/src/mito2/src/worker/handle_catchup.rs b/src/mito2/src/worker/handle_catchup.rs index c526aab8c0c..f817c1309d8 100644 --- a/src/mito2/src/worker/handle_catchup.rs +++ b/src/mito2/src/worker/handle_catchup.rs @@ -28,6 +28,8 @@ use crate::region::MitoRegion; use crate::region::catchup::RegionCatchupTask; use crate::region::opener::RegionOpener; use crate::request::OptionOutputTx; +#[cfg(feature = "hdfs-object-store")] +use crate::temp_file_sweeper; use crate::wal::entry_distributor::WalEntryReceiver; use crate::worker::RegionWorkerLoop; @@ -69,6 +71,10 @@ impl RegionWorkerLoop { let catchup_regions = self.catchup_regions.clone(); let wal = self.wal.clone(); let allow_stale_entries = self.config.allow_stale_entries; + #[cfg(feature = "hdfs-object-store")] + let sweep_older_than = self.config.gc.unknown_file_lingering_time; + #[cfg(feature = "hdfs-object-store")] + let sweep_limiter = self.sweep_limiter.clone(); common_runtime::spawn_global(async move { let mut task = RegionCatchupTask::new(region.clone(), wal, allow_stale_entries) .with_entry_receiver(entry_receiver) @@ -80,11 +86,22 @@ impl RegionWorkerLoop { if request.set_writable { region.set_role(RegionRole::Leader); // Finalize leadership: persist backfilled metadata. - if let Err(err) = region + match region .set_role_state_gracefully(SettableRegionRoleState::Leader) .await { - error!(err; "Failed to set region {region_id} to leader"); + Ok(()) => { + #[cfg(feature = "hdfs-object-store")] + temp_file_sweeper::spawn_region_sweep( + region.access_layer(), + region_id, + sweep_older_than, + sweep_limiter, + ); + } + Err(err) => { + error!(err; "Failed to set region {region_id} to leader"); + } } } catchup_regions.remove_region(region_id); diff --git a/src/mito2/src/worker/handle_create.rs b/src/mito2/src/worker/handle_create.rs index cff77b50f5f..c31e9d4ac7c 100644 --- a/src/mito2/src/worker/handle_create.rs +++ b/src/mito2/src/worker/handle_create.rs @@ -22,6 +22,8 @@ use store_api::storage::RegionId; use crate::error::Result; use crate::region::opener::{RegionOpener, check_recovered_region}; +#[cfg(feature = "hdfs-object-store")] +use crate::temp_file_sweeper; use crate::worker::RegionWorkerLoop; impl RegionWorkerLoop { @@ -79,6 +81,16 @@ impl RegionWorkerLoop { let region = opener.create_or_open(&self.config, &self.wal).await?; + // A Create retried after a crash opens an existing region as a leader + // without a role-state change, so sweep here as well. + #[cfg(feature = "hdfs-object-store")] + temp_file_sweeper::spawn_region_sweep( + region.access_layer(), + region_id, + self.config.gc.unknown_file_lingering_time, + self.sweep_limiter.clone(), + ); + info!( "A new region created with requirement {:?}, worker: {}, region: {:?}", requirements, diff --git a/src/object-store/src/factory.rs b/src/object-store/src/factory.rs index 356d3aecb21..e9dc86aa96b 100644 --- a/src/object-store/src/factory.rs +++ b/src/object-store/src/factory.rs @@ -197,6 +197,8 @@ mod tests { let store = new_hdfs_object_store(&config).await.unwrap(); assert_eq!(HDFS_NATIVE_SCHEME, store.info().scheme()); assert_eq!("/greptimedb/", store.info().root()); + // The sweep recognizes this store as one whose writes leave temporary files. + assert!(crate::temp_file::writes_temporary_files(&store)); } #[tokio::test] diff --git a/src/object-store/src/layers/hdfs.rs b/src/object-store/src/layers/hdfs.rs index ed8b2019450..529c137cdff 100644 --- a/src/object-store/src/layers/hdfs.rs +++ b/src/object-store/src/layers/hdfs.rs @@ -23,7 +23,8 @@ use opendal::raw::{ OpWrite, RpCreateDir, RpPresign, RpRename, RpStat, Service, ServiceInfo, Servicer, oio, }; use opendal::{Buffer, Capability, ErrorKind, Metadata, OperationContext, Result}; -use uuid::Uuid; + +use crate::temp_file::temporary_path; /// Adds atomic writes and streaming copies to the native HDFS backend. #[derive(Debug, Clone)] @@ -219,7 +220,10 @@ impl oio::Write for HdfsWriter { Ok(metadata) => metadata, Err(error) => { drop(writer); - let _ = delete_path(inner, context, temporary_path).await; + if let Err(delete_error) = delete_path(inner, context, temporary_path).await + { + warn!(delete_error; "Failed to delete temporary file {temporary_path} after a failed HDFS close"); + } return Err(error); } }; @@ -229,7 +233,9 @@ impl oio::Write for HdfsWriter { .rename(inner, context, temporary_path, target_path) .await { - let _ = delete_path(inner, context, temporary_path).await; + if let Err(delete_error) = delete_path(inner, context, temporary_path).await { + warn!(delete_error; "Failed to delete temporary file {temporary_path} after a failed HDFS rename"); + } return Err(error); } @@ -424,7 +430,9 @@ async fn copy_via_read_write( Ok(metadata) => metadata, Err(error) => { drop(writer); - let _ = delete_path(&inner, context, &temporary_path).await; + if let Err(delete_error) = delete_path(&inner, context, &temporary_path).await { + warn!(delete_error; "Failed to delete temporary file {temporary_path} after a failed HDFS close"); + } return Err(error); } }; @@ -434,7 +442,9 @@ async fn copy_via_read_write( .rename(&inner, context, &temporary_path, target_path) .await { - let _ = delete_path(&inner, context, &temporary_path).await; + if let Err(delete_error) = delete_path(&inner, context, &temporary_path).await { + warn!(delete_error; "Failed to delete temporary file {temporary_path} after a failed HDFS rename"); + } return Err(error); } @@ -462,15 +472,6 @@ async fn delete_path(inner: &Servicer, context: &OperationContext, path: &str) - deleter.close().await } -// TODO(fengjiachun): Clean up temporary files left behind after a process crash. -fn temporary_path(path: &str) -> String { - let suffix = format!(".greptime-{}.tmp", Uuid::new_v4()); - match path.rsplit_once('/') { - Some((parent, name)) => format!("{parent}/.{name}{suffix}"), - None => format!(".{path}{suffix}"), - } -} - #[cfg(test)] mod tests { use opendal::services::Fs; @@ -565,6 +566,25 @@ mod tests { ); } + #[tokio::test] + async fn test_list_reports_temporary_file_of_crashed_writer() { + // Regression test for #9591: a temporary file left by a crashed writer + // must stay visible to a sweep over the directory of its target. + let (_directory, store) = test_store(); + store.create_dir("data/").await.unwrap(); + let temporary = temporary_path("data/manifest.json"); + assert!(temporary.starts_with("data/.manifest.json.greptime-")); + assert!(temporary.ends_with(".tmp")); + store.write(&temporary, "partial").await.unwrap(); + + let entries = store.list("data/").await.unwrap(); + let entry = entries + .iter() + .find(|entry| entry.path() == temporary) + .unwrap(); + assert!(entry.metadata().last_modified().is_some()); + } + #[tokio::test] async fn test_copy_fallback_streams_to_target() { let (_directory, store) = test_store(); diff --git a/src/object-store/src/lib.rs b/src/object-store/src/lib.rs index 6c8956445d8..e1c421797da 100644 --- a/src/object-store/src/lib.rs +++ b/src/object-store/src/lib.rs @@ -25,6 +25,8 @@ pub mod factory; pub mod layers; pub mod manager; pub mod secure_fs; +#[cfg(feature = "hdfs-object-store")] +pub mod temp_file; #[cfg(feature = "testing")] pub mod test_util; pub mod util; diff --git a/src/object-store/src/temp_file.rs b/src/object-store/src/temp_file.rs new file mode 100644 index 00000000000..ee169cbeefe --- /dev/null +++ b/src/object-store/src/temp_file.rs @@ -0,0 +1,373 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Temporary files of non-append writes, and how to reclaim them. + +use std::time::Duration; + +use common_telemetry::warn; +use opendal::raw::Timestamp; +use uuid::Uuid; + +use crate::{ErrorKind, ObjectStore}; + +/// Prefix of a temporary file name. +const TEMPORARY_PREFIX: &str = "."; + +/// Marker that follows the name of the target in a temporary file name. +const TEMPORARY_MARKER: &str = ".greptime-"; + +/// Suffix of a temporary file name. +const TEMPORARY_SUFFIX: &str = ".tmp"; + +/// Scheme of the only service that publishes non-append writes with a rename. +const TEMPORARY_FILE_SCHEME: &str = crate::services::HDFS_NATIVE_SCHEME; + +/// Returns true if `store` publishes non-append writes with a rename, which is +/// what leaves temporary files behind when the writer dies. +/// +/// Only the HDFS compatibility layer does that; a write to any other service +/// appears under its final name. +pub fn writes_temporary_files(store: &ObjectStore) -> bool { + store.info().scheme() == TEMPORARY_FILE_SCHEME +} + +/// What a sweep did. +/// +/// A sweep goes through its directories and files one at a time, and a failure +/// on either doesn't stop it, so it reports what it reclaimed and what it +/// couldn't. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct SweepReport { + /// Temporary files deleted. + pub deleted: usize, + /// List and delete operations that failed. + pub failed: usize, +} + +impl SweepReport { + fn merge(&mut self, other: SweepReport) { + self.deleted += other.deleted; + self.failed += other.failed; + } +} + +/// Deletes the temporary files in `dirs` that are older than `older_than`. +/// +/// A non-append write goes to a temporary file next to its target and is +/// renamed on close, so a writer that dies in between leaves the file behind +/// with nothing to reclaim it: GC can't parse the name, and the startup cleanup +/// of [`ATOMIC_WRITE_DIR`](crate::ATOMIC_WRITE_DIR) doesn't reach it. +/// +/// The caller runs this for a region that just became its writable leader, so a +/// file that is old enough belongs to a writer that is gone. Failures are +/// logged and don't stop the sweep. +pub async fn sweep_temporary_files( + store: &ObjectStore, + dirs: &[String], + older_than: Duration, +) -> SweepReport { + let mut report = SweepReport::default(); + for dir in dirs { + report.merge(sweep_dir(store, dir, older_than).await); + } + report +} + +async fn sweep_dir(store: &ObjectStore, dir: &str, older_than: Duration) -> SweepReport { + let entries = match store.list(dir).await { + Ok(entries) => entries, + // Not every region has every directory, for example one that was never + // repartitioned has no staging directory. + Err(error) if error.kind() == ErrorKind::NotFound => return SweepReport::default(), + Err(error) => { + warn!(error; "Failed to list {dir} for temporary files"); + return SweepReport { + failed: 1, + ..Default::default() + }; + } + }; + + let expired_before = Timestamp::now() + .into_inner() + .as_millisecond() + .saturating_sub(i64::try_from(older_than.as_millis()).unwrap_or(i64::MAX)); + + let mut report = SweepReport::default(); + for entry in entries { + if !is_temporary_file(entry.name()) { + continue; + } + + // HDFS advances the modification time of a file under construction only + // as its blocks complete, so an old file is not necessarily abandoned. + // That is what the threshold covers. + let is_expired = entry + .metadata() + .last_modified() + .map(|last_modified| last_modified.into_inner().as_millisecond() < expired_before) + // Without a modification time there is no way to tell whether the + // writer is gone, so keep the file. + .unwrap_or(false); + if !is_expired { + continue; + } + + // Delete one file at a time: the bulk delete stops at the first + // failure, and one file that can't be deleted must not keep the rest of + // the directory from being swept. + let path = entry.path(); + match store.delete(path).await { + Ok(()) => report.deleted += 1, + Err(error) => { + warn!(error; "Failed to delete temporary file {path}"); + report.failed += 1; + } + } + } + + report +} + +/// Returns the path a non-append write to `path` goes to before it is renamed. +/// +/// A writer that dies in between leaves the file behind, which is what +/// [`sweep_temporary_files`] reclaims. +pub(crate) fn temporary_path(path: &str) -> String { + let suffix = format!("{TEMPORARY_MARKER}{}{TEMPORARY_SUFFIX}", Uuid::new_v4()); + match path.rsplit_once('/') { + Some((parent, name)) => format!("{parent}/{TEMPORARY_PREFIX}{name}{suffix}"), + None => format!("{TEMPORARY_PREFIX}{path}{suffix}"), + } +} + +fn is_temporary_file(name: &str) -> bool { + // `name` is the last segment of a path, so a directory name ends with `/`. + name.starts_with(TEMPORARY_PREFIX) + && name.contains(TEMPORARY_MARKER) + && name.ends_with(TEMPORARY_SUFFIX) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use opendal::Operator; + use opendal::services::Fs; + use tempfile::TempDir; + + use super::*; + use crate::layers::HdfsCompatibilityLayer; + use crate::layers::mock::{MockLayer, MockLayerBuilder, OpDelete, oio}; + use crate::{Error, Result}; + + const EXPIRED: Duration = Duration::from_millis(1); + + fn test_store() -> (TempDir, ObjectStore) { + let directory = tempfile::tempdir().unwrap(); + let store = Operator::new(Fs::default().root(directory.path().to_str().unwrap())).unwrap(); + (directory, store) + } + + fn test_store_of(layer: MockLayer) -> (TempDir, ObjectStore) { + let directory = tempfile::tempdir().unwrap(); + let store = Operator::new(Fs::default().root(directory.path().to_str().unwrap())) + .unwrap() + .layer(layer); + (directory, store) + } + + fn test_store_of_failing_list(kind: ErrorKind) -> (TempDir, ObjectStore) { + struct FailingLister(ErrorKind); + + impl oio::List for FailingLister { + async fn next(&mut self) -> Result> { + Err(Error::new(self.0, "Failed to list")) + } + } + + let layer = MockLayerBuilder::default() + .lister_factory(Arc::new(move |_, _, _| Box::new(FailingLister(kind)))) + .build() + .unwrap(); + test_store_of(layer) + } + + // A store that fails to delete every path containing `marker`. + fn test_store_of_failing_delete(marker: &'static str) -> (TempDir, ObjectStore) { + struct FailingDeleter { + inner: oio::Deleter, + marker: &'static str, + } + + impl oio::Delete for FailingDeleter { + async fn delete(&mut self, path: &str, args: OpDelete) -> Result<()> { + if path.contains(self.marker) { + return Err(Error::new(ErrorKind::Unexpected, "Failed to delete")); + } + self.inner.delete(path, args).await + } + + async fn close(&mut self) -> Result<()> { + self.inner.close().await + } + } + + let layer = MockLayerBuilder::default() + .deleter_factory(Arc::new(move |inner| { + Box::new(FailingDeleter { inner, marker }) + })) + .build() + .unwrap(); + test_store_of(layer) + } + + async fn wait_until_expired() { + // A file written just before this call can carry the same modification + // time in milliseconds as the sweep that follows it, which would make the + // sweep keep it. + tokio::time::sleep(Duration::from_millis(10)).await; + } + + #[test] + fn test_is_temporary_file() { + assert!(is_temporary_file( + ".00000000000000000042.json.greptime-53f0f4a6-3f3e-4a2f-9a2f-6fe4b0d5b5b0.tmp" + )); + assert!(is_temporary_file(".a5f2.parquet.greptime-53f0f4a6.tmp")); + + assert!(!is_temporary_file("00000000000000000042.json")); + assert!(!is_temporary_file(".00000000000000000042.json")); + assert!(!is_temporary_file( + "00000000000000000042.json.greptime-53f0f4a6.tmp" + )); + assert!(!is_temporary_file( + ".00000000000000000042.json.greptime-53f0f4a6" + )); + assert!(!is_temporary_file("index/")); + } + + #[tokio::test] + async fn test_sweep_dir_deletes_expired_temporary_file() { + let (_directory, store) = test_store(); + let temporary = "data/.42.parquet.greptime-53f0f4a6.tmp"; + store.create_dir("data/").await.unwrap(); + store.write(temporary, "partial").await.unwrap(); + store.write("data/42.parquet", "data").await.unwrap(); + + wait_until_expired().await; + assert_eq!( + SweepReport { + deleted: 1, + failed: 0 + }, + sweep_dir(&store, "data/", EXPIRED).await + ); + + assert!(!store.exists(temporary).await.unwrap()); + assert!(store.exists("data/42.parquet").await.unwrap()); + } + + #[tokio::test] + async fn test_sweep_dir_keeps_young_temporary_file() { + let (_directory, store) = test_store(); + let temporary = "data/.42.parquet.greptime-53f0f4a6.tmp"; + store.create_dir("data/").await.unwrap(); + store.write(temporary, "partial").await.unwrap(); + + assert_eq!( + SweepReport::default(), + sweep_dir(&store, "data/", Duration::from_secs(24 * 60 * 60)).await + ); + + assert!(store.exists(temporary).await.unwrap()); + } + + #[tokio::test] + async fn test_sweep_dir_of_missing_directory() { + // Fs lists a directory it doesn't have as an empty one, but HDFS reports + // it as not found. + let (_directory, store) = test_store_of_failing_list(ErrorKind::NotFound); + + // A region that was never repartitioned has no staging directory, so a + // missing directory is an expected answer rather than a failure. + assert_eq!( + SweepReport::default(), + sweep_dir(&store, "data/staging/blob/", Duration::ZERO).await + ); + } + + #[tokio::test] + async fn test_sweep_dir_reports_failed_list() { + let (_directory, store) = test_store_of_failing_list(ErrorKind::Unexpected); + + assert_eq!( + SweepReport { + deleted: 0, + failed: 1 + }, + sweep_dir(&store, "data/", Duration::ZERO).await + ); + } + + #[tokio::test] + async fn test_sweep_dir_keeps_going_after_failed_delete() { + let (_directory, store) = test_store_of_failing_delete("fail"); + let failing = "data/.fail-42.parquet.greptime-53f0f4a6.tmp"; + let sweepable = "data/.ok-42.parquet.greptime-53f0f4a6.tmp"; + store.create_dir("data/").await.unwrap(); + store.write(failing, "partial").await.unwrap(); + store.write(sweepable, "partial").await.unwrap(); + + wait_until_expired().await; + assert_eq!( + SweepReport { + deleted: 1, + failed: 1 + }, + sweep_dir(&store, "data/", EXPIRED).await + ); + + assert!(store.exists(failing).await.unwrap()); + assert!(!store.exists(sweepable).await.unwrap()); + } + + #[tokio::test] + async fn test_sweep_temporary_files_on_compatibility_layer() { + let (_directory, store) = test_store(); + let temporary = "data/.42.parquet.greptime-53f0f4a6.tmp"; + store.create_dir("data/").await.unwrap(); + // What a writer that died before it renamed its file leaves behind. + store.write(temporary, "partial").await.unwrap(); + // The same layer HDFS writes through, on an Fs store whose scheme stays `fs`. + let store = store.layer(HdfsCompatibilityLayer::new_for_test()); + + wait_until_expired().await; + assert_eq!( + SweepReport { + deleted: 1, + failed: 0 + }, + sweep_temporary_files( + &store, + &["data/".to_string(), "data/staging/blob/".to_string()], + EXPIRED, + ) + .await + ); + + assert!(!store.exists(temporary).await.unwrap()); + } +}