mito2/series_index/
purger.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//! Deferred deletion of aggregate series-index files.
16
17use std::fmt::{self, Debug, Formatter};
18
19use common_telemetry::{info, warn};
20use object_store::{ErrorKind, ObjectStore};
21use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel};
22
23use crate::metrics::SERIES_INDEX_FILE_OPERATION_TOTAL;
24use crate::series_index::catalog::series_index_path;
25use crate::sst::file::RegionFileId;
26
27#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
28pub(crate) enum IndexFileType {
29    Range,
30    Series,
31}
32
33impl IndexFileType {
34    fn as_str(self) -> &'static str {
35        match self {
36            Self::Range => "range",
37            Self::Series => "series",
38        }
39    }
40}
41
42#[derive(Debug, Clone, Copy)]
43pub(crate) struct PurgeRequest {
44    pub(crate) file_id: RegionFileId,
45}
46
47#[derive(Clone)]
48pub(crate) struct IndexFilePurger {
49    store: ObjectStore,
50    sender: UnboundedSender<PurgeRequest>,
51}
52
53impl Debug for IndexFilePurger {
54    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
55        f.debug_struct("IndexFilePurger").finish_non_exhaustive()
56    }
57}
58
59impl IndexFilePurger {
60    pub(crate) fn purge(&self, request: PurgeRequest) {
61        if let Err(error) = self.sender.send(request) {
62            let store = self.store.clone();
63            common_runtime::spawn_global(async move {
64                purge_file(&store, error.0).await;
65            });
66        }
67    }
68}
69
70pub(crate) fn file_operation(index_type: IndexFileType, operation: &str, result: &str) {
71    SERIES_INDEX_FILE_OPERATION_TOTAL
72        .with_label_values(&[index_type.as_str(), operation, result])
73        .inc();
74}
75
76/// Processes queued deletions once each, independently of periodic maintenance.
77pub(crate) async fn run_index_purge_task(
78    worker_id: u32,
79    store: ObjectStore,
80    mut receiver: UnboundedReceiver<PurgeRequest>,
81) {
82    info!("Start series-index purge task, worker: {worker_id}");
83    while let Some(request) = receiver.recv().await {
84        purge_file(&store, request).await;
85    }
86    info!("Stop series-index purge task, worker: {worker_id}");
87}
88
89async fn purge_file(store: &ObjectStore, request: PurgeRequest) {
90    let path = series_index_path(request.file_id.region_id(), request.file_id.file_id());
91    match store.delete(&path).await {
92        Ok(()) => {
93            file_operation(IndexFileType::Series, "delete", "success");
94        }
95        Err(error) if error.kind() == ErrorKind::NotFound => {
96            file_operation(IndexFileType::Series, "delete", "success");
97        }
98        Err(error) => {
99            file_operation(IndexFileType::Series, "delete", "failure");
100            warn!(error; "Failed to delete series index, index_type: {}, path: {}, phase: deletion", IndexFileType::Series.as_str(), path);
101        }
102    }
103}
104
105pub(crate) fn series_index_channel(
106    store: ObjectStore,
107) -> (IndexFilePurger, UnboundedReceiver<PurgeRequest>) {
108    let (sender, receiver) = unbounded_channel();
109    (IndexFilePurger { store, sender }, receiver)
110}
mito2/series_index/purger.rs

mito2/series_index/
purger.rs
