1use std::time::Duration;
18
19use common_base::readable_size::ReadableSize;
20use common_config::{Configurable, DEFAULT_DATA_HOME};
21use common_options::memory::MemoryOptions;
22pub use common_procedure::options::ProcedureConfig;
23use common_telemetry::logging::{LoggingOptions, TracingOptions};
24use common_wal::config::DatanodeWalConfig;
25use common_workload::{DatanodeWorkloadType, sanitize_workload_types};
26use file_engine::config::EngineConfig as FileEngineConfig;
27use meta_client::MetaClientOptions;
28use metric_engine::config::EngineConfig as MetricEngineConfig;
29use mito2::config::MitoConfig;
30pub(crate) use object_store::config::ObjectStoreConfig;
31use query::options::QueryOptions;
32use serde::{Deserialize, Serialize};
33use servers::grpc::GrpcOptions;
34use servers::http::HttpOptions;
35
36#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
38#[serde(default)]
39pub struct StorageConfig {
40 pub data_home: String,
42 pub copy_root: Option<String>,
47 #[serde(flatten)]
48 pub store: ObjectStoreConfig,
49 pub providers: Vec<ObjectStoreConfig>,
51}
52
53impl StorageConfig {
54 pub fn is_object_storage(&self) -> bool {
56 self.store.is_object_storage()
57 }
58}
59
60impl Default for StorageConfig {
61 fn default() -> Self {
62 Self {
63 data_home: DEFAULT_DATA_HOME.to_string(),
64 copy_root: None,
65 store: ObjectStoreConfig::default(),
66 providers: vec![],
67 }
68 }
69}
70
71#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
72#[serde(default)]
73pub struct DatanodeOptions {
74 pub node_id: Option<u64>,
75 pub default_column_prefix: Option<String>,
76 pub workload_types: Vec<DatanodeWorkloadType>,
77 pub require_lease_before_startup: bool,
78 pub init_regions_in_background: bool,
79 pub init_regions_parallelism: usize,
80 pub grpc: GrpcOptions,
81 pub http: HttpOptions,
82 pub meta_client: Option<MetaClientOptions>,
83 pub wal: DatanodeWalConfig,
84 pub storage: StorageConfig,
85 pub max_concurrent_queries: usize,
86 #[serde(with = "humantime_serde")]
89 pub concurrent_query_limiter_timeout: Duration,
90 pub region_engine: Vec<RegionEngineConfig>,
92 pub logging: LoggingOptions,
93 pub enable_telemetry: bool,
94 pub tracing: TracingOptions,
95 pub query: QueryOptions,
96 pub memory: MemoryOptions,
97
98 pub heartbeat_env_vars: Vec<String>,
101
102 #[deprecated(note = "Please use `grpc.bind_addr` instead.")]
104 pub rpc_addr: Option<String>,
105 #[deprecated(note = "Please use `grpc.server_addr` instead.")]
106 pub rpc_hostname: Option<String>,
107 #[deprecated(note = "Please use `grpc.runtime_size` instead.")]
108 pub rpc_runtime_size: Option<usize>,
109 #[deprecated(note = "Please use `grpc.max_recv_message_size` instead.")]
110 pub rpc_max_recv_message_size: Option<ReadableSize>,
111 #[deprecated(note = "Please use `grpc.max_send_message_size` instead.")]
112 pub rpc_max_send_message_size: Option<ReadableSize>,
113}
114
115impl DatanodeOptions {
116 pub fn sanitize(&mut self) {
118 sanitize_workload_types(&mut self.workload_types);
119
120 if self.storage.is_object_storage() {
121 self.storage
122 .store
123 .cache_config_mut()
124 .unwrap()
125 .sanitize(&self.storage.data_home);
126 }
127 }
128}
129
130impl Default for DatanodeOptions {
131 #[allow(deprecated)]
132 fn default() -> Self {
133 Self {
134 node_id: None,
135 default_column_prefix: None,
136 workload_types: vec![DatanodeWorkloadType::Hybrid],
137 require_lease_before_startup: false,
138 init_regions_in_background: false,
139 init_regions_parallelism: 16,
140 grpc: GrpcOptions::default().with_bind_addr("127.0.0.1:3001"),
141 http: HttpOptions::default(),
142 meta_client: None,
143 wal: DatanodeWalConfig::default(),
144 storage: StorageConfig::default(),
145 max_concurrent_queries: 0,
146 concurrent_query_limiter_timeout: Duration::from_millis(100),
147 region_engine: vec![
148 RegionEngineConfig::Mito(MitoConfig::default()),
149 RegionEngineConfig::File(FileEngineConfig::default()),
150 ],
151 logging: LoggingOptions::default(),
152 enable_telemetry: true,
153 tracing: TracingOptions::default(),
154 query: QueryOptions::default(),
155 memory: MemoryOptions::default(),
156 heartbeat_env_vars: vec![],
157
158 rpc_addr: None,
160 rpc_hostname: None,
161 rpc_runtime_size: None,
162 rpc_max_recv_message_size: None,
163 rpc_max_send_message_size: None,
164 }
165 }
166}
167
168impl Configurable for DatanodeOptions {
169 fn env_list_keys() -> Option<&'static [&'static str]> {
170 Some(&[
171 "heartbeat_env_vars",
172 "meta_client.metasrv_addrs",
173 "wal.broker_endpoints",
174 ])
175 }
176}
177
178#[allow(clippy::large_enum_variant)]
179#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
180pub enum RegionEngineConfig {
181 #[serde(rename = "mito")]
182 Mito(MitoConfig),
183 #[serde(rename = "file")]
184 File(FileEngineConfig),
185 #[serde(rename = "metric")]
186 Metric(MetricEngineConfig),
187}
188
189#[cfg(test)]
190mod tests {
191 use common_base::secrets::ExposeSecret;
192
193 use super::*;
194
195 #[test]
196 fn test_toml() {
197 let opts = DatanodeOptions::default();
198 let toml_string = toml::to_string(&opts).unwrap();
199 let _parsed: DatanodeOptions = toml::from_str(&toml_string).unwrap();
200 }
201
202 #[test]
203 fn test_secstr() {
204 let toml_str = r#"
205 [storage]
206 type = "S3"
207 access_key_id = "access_key_id"
208 secret_access_key = "secret_access_key"
209 "#;
210 let opts: DatanodeOptions = toml::from_str(toml_str).unwrap();
211 match &opts.storage.store {
212 ObjectStoreConfig::S3(cfg) => {
213 assert_eq!(
214 "SecretBox<alloc::string::String>([REDACTED])".to_string(),
215 format!("{:?}", cfg.connection.access_key_id)
216 );
217 assert_eq!(
218 "access_key_id",
219 cfg.connection.access_key_id.expose_secret()
220 );
221 }
222 _ => unreachable!(),
223 }
224 }
225 #[test]
226 fn test_skip_ssl_validation_config() {
227 let toml_str_true = r#"
229 [storage]
230 type = "S3"
231 [storage.http_client]
232 skip_ssl_validation = true
233 "#;
234 let opts: DatanodeOptions = toml::from_str(toml_str_true).unwrap();
235 match &opts.storage.store {
236 ObjectStoreConfig::S3(cfg) => {
237 assert!(cfg.http_client.skip_ssl_validation);
238 }
239 _ => panic!("Expected S3 config"),
240 }
241
242 let toml_str_false = r#"
244 [storage]
245 type = "S3"
246 [storage.http_client]
247 skip_ssl_validation = false
248 "#;
249 let opts: DatanodeOptions = toml::from_str(toml_str_false).unwrap();
250 match &opts.storage.store {
251 ObjectStoreConfig::S3(cfg) => {
252 assert!(!cfg.http_client.skip_ssl_validation);
253 }
254 _ => panic!("Expected S3 config"),
255 }
256 let toml_str_default = r#"
258 [storage]
259 type = "S3"
260 "#;
261 let opts: DatanodeOptions = toml::from_str(toml_str_default).unwrap();
262 match &opts.storage.store {
263 ObjectStoreConfig::S3(cfg) => {
264 assert!(!cfg.http_client.skip_ssl_validation);
265 }
266 _ => panic!("Expected S3 config"),
267 }
268 }
269
270 #[test]
271 fn test_cache_config() {
272 let toml_str = r#"
273 [storage]
274 data_home = "test_data_home"
275 type = "S3"
276 [storage.cache_config]
277 enable_read_cache = true
278 "#;
279 let mut opts: DatanodeOptions = toml::from_str(toml_str).unwrap();
280 opts.sanitize();
281 assert!(opts.storage.store.cache_config().unwrap().enable_read_cache);
282 assert_eq!(
283 opts.storage.store.cache_config().unwrap().cache_path,
284 "test_data_home"
285 );
286 }
287}