mito2/
temp_file_sweeper.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Reclaims the temporary files that a crashed writer left in a region.
16
17use std::sync::Arc;
18use std::time::Duration;
19
20use common_telemetry::info;
21use object_store::temp_file::{SweepReport, sweep_temporary_files, writes_temporary_files};
22use store_api::storage::RegionId;
23use tokio::sync::Semaphore;
24
25use crate::access_layer::{AccessLayer, AccessLayerRef};
26use crate::manifest::storage::manifest_dir;
27use crate::manifest::storage::staging::{staging_blob_path, staging_manifest_path};
28use crate::metrics::{SWEEP_ERRORS_TOTAL, SWEEP_FILES_DELETED_TOTAL, SWEEP_RUNS_TOTAL};
29use crate::sst::location::index_dir;
30
31/// Returns the directories that `region_id` writes non-append files into.
32///
33/// Every one of them can hold the temporary file of a crashed writer, and a
34/// list is not recursive, so the sweep has to name each directory.
35pub(crate) fn region_write_dirs(access_layer: &AccessLayer, region_id: RegionId) -> Vec<String> {
36    let region_dir = access_layer.build_region_dir(region_id);
37    let manifest_path = manifest_dir(&region_dir);
38    vec![
39        region_dir.clone(),
40        index_dir(&region_dir),
41        manifest_path.clone(),
42        staging_blob_path(&manifest_path),
43        staging_manifest_path(&manifest_path),
44    ]
45}
46
47/// Sweeps the temporary files of `region_id` in the background.
48///
49/// The caller is on the path that makes the region its writable leader, so
50/// reclaiming leftovers must not slow it down or fail it: everything happens in
51/// a spawned task that only logs.
52///
53/// `older_than` is the margin the caller takes from GC's
54/// `unknown_file_lingering_time`, which guards the same thing: a file this old
55/// is not one a writer is still filling.
56///
57/// A sweep waits for a permit of `sweep_limiter`, so a `None` means the
58/// datanode allows no concurrent GC job and there is no sweep.
59pub(crate) fn spawn_region_sweep(
60    access_layer: AccessLayerRef,
61    region_id: RegionId,
62    older_than: Duration,
63    sweep_limiter: Option<Arc<Semaphore>>,
64) {
65    // No write to this store leaves a temporary file behind, so don't spend a
66    // task on it: a sweep would still list every directory of the region to
67    // find that out.
68    if !writes_temporary_files(access_layer.object_store()) {
69        return;
70    }
71
72    let Some(sweep_limiter) = sweep_limiter else {
73        return;
74    };
75
76    common_runtime::spawn_global(async move {
77        // A burst of regions becoming writable must not run more sweeps at once
78        // than the datanode allows GC jobs. Wait for a permit instead of giving
79        // up, the sweep has no deadline.
80        let Ok(_permit) = sweep_limiter.acquire_owned().await else {
81            return;
82        };
83
84        let dirs = region_write_dirs(&access_layer, region_id);
85        let report = sweep_temporary_files(access_layer.object_store(), &dirs, older_than).await;
86        record_sweep(&report);
87        if report.deleted > 0 {
88            info!(
89                "Reclaimed {} temporary file(s) of a crashed writer in region {region_id}",
90                report.deleted
91            );
92        }
93    });
94}
95
96fn record_sweep(report: &SweepReport) {
97    SWEEP_RUNS_TOTAL.inc();
98    SWEEP_FILES_DELETED_TOTAL.inc_by(report.deleted as u64);
99    SWEEP_ERRORS_TOTAL.inc_by(report.failed as u64);
100}
101
102#[cfg(test)]
103mod tests {
104    use object_store::ObjectStore;
105    use object_store::services::Memory;
106    use store_api::region_request::PathType;
107
108    use super::*;
109    use crate::test_util::TestEnv;
110
111    fn test_access_layer(env: &TestEnv) -> AccessLayer {
112        AccessLayer::new(
113            "test",
114            PathType::Bare,
115            ObjectStore::new(Memory::default()).unwrap(),
116            env.get_puffin_manager(),
117            env.get_intermediate_manager(),
118        )
119    }
120
121    #[tokio::test]
122    async fn test_region_write_dirs() {
123        let env = TestEnv::new().await;
124        let access_layer = test_access_layer(&env);
125        let region_id = RegionId::new(1, 2);
126
127        let dirs = region_write_dirs(&access_layer, region_id);
128
129        assert_eq!(5, dirs.len());
130        assert!(dirs[0].ends_with("1_0000000002/"));
131        assert!(dirs[1].ends_with("1_0000000002/index/"));
132        assert!(dirs[2].ends_with("1_0000000002/manifest/"));
133        assert!(dirs[3].ends_with("1_0000000002/staging/blob/"));
134        assert!(dirs[4].ends_with("1_0000000002/staging/manifest/"));
135    }
136
137    #[test]
138    fn test_record_sweep() {
139        let runs = SWEEP_RUNS_TOTAL.get();
140        let deleted = SWEEP_FILES_DELETED_TOTAL.get();
141        let errors = SWEEP_ERRORS_TOTAL.get();
142
143        record_sweep(&SweepReport {
144            deleted: 2,
145            failed: 1,
146        });
147
148        assert_eq!(runs + 1, SWEEP_RUNS_TOTAL.get());
149        assert_eq!(deleted + 2, SWEEP_FILES_DELETED_TOTAL.get());
150        assert_eq!(errors + 1, SWEEP_ERRORS_TOTAL.get());
151    }
152}
mito2/temp_file_sweeper.rs

mito2/
temp_file_sweeper.rs
