metric_engine/
repeated_task.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
15use std::sync::{Arc, RwLock};
16use std::time::Instant;
17
18use common_runtime::TaskFunction;
19use common_telemetry::{debug, error};
20use mito2::engine::MitoEngine;
21use store_api::region_engine::{RegionEngine, RegionRole};
22use store_api::region_request::{RegionFlushRequest, RegionRequest};
23
24use crate::engine::MetricEngineState;
25use crate::error::{Error, Result};
26use crate::utils;
27
28/// Task to flush metadata regions.
29///
30/// This task is used to send flush requests to the metadata regions
31/// periodically.
32pub(crate) struct FlushMetadataRegionTask {
33    pub(crate) state: Arc<RwLock<MetricEngineState>>,
34    pub(crate) mito: MitoEngine,
35}
36
37#[async_trait::async_trait]
38impl TaskFunction<Error> for FlushMetadataRegionTask {
39    fn name(&self) -> &str {
40        "FlushMetadataRegionTask"
41    }
42
43    async fn call(&mut self) -> Result<()> {
44        let region_ids = {
45            let state = self.state.read().unwrap();
46            state
47                .physical_region_states()
48                .keys()
49                .cloned()
50                .collect::<Vec<_>>()
51        };
52
53        let num_region = region_ids.len();
54        let now = Instant::now();
55        for region_id in region_ids {
56            let Some(role) = self.mito.role(region_id) else {
57                continue;
58            };
59            if role == RegionRole::Follower {
60                continue;
61            }
62            let metadata_region_id = utils::to_metadata_region_id(region_id);
63            if let Err(e) = self
64                .mito
65                .handle_request(
66                    metadata_region_id,
67                    RegionRequest::Flush(RegionFlushRequest::default()),
68                )
69                .await
70            {
71                error!(e; "Failed to flush metadata region {}", metadata_region_id);
72            }
73        }
74        debug!(
75            "Flushed {} metadata regions, elapsed: {:?}",
76            num_region,
77            now.elapsed()
78        );
79
80        Ok(())
81    }
82}
83
84#[cfg(test)]
85mod tests {
86    use std::assert_matches;
87    use std::time::Duration;
88
89    use store_api::region_engine::{RegionEngine, RegionManifestInfo};
90
91    use crate::config::{DEFAULT_FLUSH_METADATA_REGION_INTERVAL, EngineConfig};
92    use crate::test_util::TestEnv;
93
94    #[tokio::test]
95    async fn test_flush_metadata_region_task() {
96        let env = TestEnv::with_prefix_and_config(
97            "test_flush_metadata_region_task",
98            EngineConfig {
99                flush_metadata_region_interval: Duration::from_millis(10),
100            },
101        )
102        .await;
103        env.init_metric_region().await;
104        let engine = env.metric();
105        // Wait for flush task run
106        tokio::time::sleep(Duration::from_millis(500)).await;
107        let physical_region_id = env.default_physical_region_id();
108        let stat = engine.region_statistic(physical_region_id).unwrap();
109
110        assert_matches!(
111            stat.manifest,
112            RegionManifestInfo::Metric {
113                metadata_manifest_version: 1,
114                metadata_flushed_entry_id: 1,
115                ..
116            }
117        )
118    }
119
120    #[tokio::test]
121    async fn test_flush_metadata_region_task_with_long_interval() {
122        let env = TestEnv::with_prefix_and_config(
123            "test_flush_metadata_region_task_with_long_interval",
124            EngineConfig {
125                flush_metadata_region_interval: Duration::from_secs(60),
126            },
127        )
128        .await;
129        env.init_metric_region().await;
130        let engine = env.metric();
131        // Wait for flush task run, should not flush metadata region
132        tokio::time::sleep(Duration::from_millis(200)).await;
133        let physical_region_id = env.default_physical_region_id();
134        let stat = engine.region_statistic(physical_region_id).unwrap();
135
136        assert_matches!(
137            stat.manifest,
138            RegionManifestInfo::Metric {
139                metadata_manifest_version: 0,
140                metadata_flushed_entry_id: 0,
141                ..
142            }
143        )
144    }
145
146    #[test]
147    fn test_flush_metadata_region_sanitize() {
148        let mut config = EngineConfig {
149            flush_metadata_region_interval: Duration::from_secs(0),
150        };
151        config.sanitize();
152        assert_eq!(
153            config.flush_metadata_region_interval,
154            DEFAULT_FLUSH_METADATA_REGION_INTERVAL
155        );
156    }
157}
metric_engine/repeated_task.rs

metric_engine/
repeated_task.rs
