1use common_base::readable_size::ReadableSize;
18use common_config::{Configurable, DEFAULT_DATA_HOME};
19use common_options::memory::MemoryOptions;
20pub use common_procedure::options::ProcedureConfig;
21use common_telemetry::logging::{LoggingOptions, TracingOptions};
22use common_wal::config::DatanodeWalConfig;
23use common_workload::{sanitize_workload_types, DatanodeWorkloadType};
24use file_engine::config::EngineConfig as FileEngineConfig;
25use meta_client::MetaClientOptions;
26use metric_engine::config::EngineConfig as MetricEngineConfig;
27use mito2::config::MitoConfig;
28pub(crate) use object_store::config::ObjectStoreConfig;
29use query::options::QueryOptions;
30use serde::{Deserialize, Serialize};
31use servers::export_metrics::ExportMetricsOption;
32use servers::grpc::GrpcOptions;
33use servers::heartbeat_options::HeartbeatOptions;
34use servers::http::HttpOptions;
35
36pub const DEFAULT_OBJECT_STORE_CACHE_SIZE: ReadableSize = ReadableSize::gb(5);
37
38#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
40#[serde(default)]
41pub struct StorageConfig {
42 pub data_home: String,
44 #[serde(flatten)]
45 pub store: ObjectStoreConfig,
46 pub providers: Vec<ObjectStoreConfig>,
48}
49
50impl StorageConfig {
51 pub fn is_object_storage(&self) -> bool {
53 self.store.is_object_storage()
54 }
55}
56
57impl Default for StorageConfig {
58 fn default() -> Self {
59 Self {
60 data_home: DEFAULT_DATA_HOME.to_string(),
61 store: ObjectStoreConfig::default(),
62 providers: vec![],
63 }
64 }
65}
66
67#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
68#[serde(default)]
69pub struct DatanodeOptions {
70 pub node_id: Option<u64>,
71 pub workload_types: Vec<DatanodeWorkloadType>,
72 pub require_lease_before_startup: bool,
73 pub init_regions_in_background: bool,
74 pub init_regions_parallelism: usize,
75 pub grpc: GrpcOptions,
76 pub heartbeat: HeartbeatOptions,
77 pub http: HttpOptions,
78 pub meta_client: Option<MetaClientOptions>,
79 pub wal: DatanodeWalConfig,
80 pub storage: StorageConfig,
81 pub max_concurrent_queries: usize,
82 pub region_engine: Vec<RegionEngineConfig>,
84 pub logging: LoggingOptions,
85 pub enable_telemetry: bool,
86 pub export_metrics: ExportMetricsOption,
87 pub tracing: TracingOptions,
88 pub query: QueryOptions,
89 pub memory: MemoryOptions,
90
91 #[deprecated(note = "Please use `grpc.addr` instead.")]
93 pub rpc_addr: Option<String>,
94 #[deprecated(note = "Please use `grpc.hostname` instead.")]
95 pub rpc_hostname: Option<String>,
96 #[deprecated(note = "Please use `grpc.runtime_size` instead.")]
97 pub rpc_runtime_size: Option<usize>,
98 #[deprecated(note = "Please use `grpc.max_recv_message_size` instead.")]
99 pub rpc_max_recv_message_size: Option<ReadableSize>,
100 #[deprecated(note = "Please use `grpc.max_send_message_size` instead.")]
101 pub rpc_max_send_message_size: Option<ReadableSize>,
102}
103
104impl DatanodeOptions {
105 pub fn sanitize(&mut self) {
107 sanitize_workload_types(&mut self.workload_types);
108 }
109}
110
111impl Default for DatanodeOptions {
112 #[allow(deprecated)]
113 fn default() -> Self {
114 Self {
115 node_id: None,
116 workload_types: vec![DatanodeWorkloadType::Hybrid],
117 require_lease_before_startup: false,
118 init_regions_in_background: false,
119 init_regions_parallelism: 16,
120 grpc: GrpcOptions::default().with_bind_addr("127.0.0.1:3001"),
121 http: HttpOptions::default(),
122 meta_client: None,
123 wal: DatanodeWalConfig::default(),
124 storage: StorageConfig::default(),
125 max_concurrent_queries: 0,
126 region_engine: vec![
127 RegionEngineConfig::Mito(MitoConfig::default()),
128 RegionEngineConfig::File(FileEngineConfig::default()),
129 ],
130 logging: LoggingOptions::default(),
131 heartbeat: HeartbeatOptions::datanode_default(),
132 enable_telemetry: true,
133 export_metrics: ExportMetricsOption::default(),
134 tracing: TracingOptions::default(),
135 query: QueryOptions::default(),
136 memory: MemoryOptions::default(),
137
138 rpc_addr: None,
140 rpc_hostname: None,
141 rpc_runtime_size: None,
142 rpc_max_recv_message_size: None,
143 rpc_max_send_message_size: None,
144 }
145 }
146}
147
148impl Configurable for DatanodeOptions {
149 fn env_list_keys() -> Option<&'static [&'static str]> {
150 Some(&["meta_client.metasrv_addrs", "wal.broker_endpoints"])
151 }
152}
153
154#[allow(clippy::large_enum_variant)]
155#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
156pub enum RegionEngineConfig {
157 #[serde(rename = "mito")]
158 Mito(MitoConfig),
159 #[serde(rename = "file")]
160 File(FileEngineConfig),
161 #[serde(rename = "metric")]
162 Metric(MetricEngineConfig),
163}
164
165#[cfg(test)]
166mod tests {
167 use common_base::secrets::ExposeSecret;
168
169 use super::*;
170
171 #[test]
172 fn test_toml() {
173 let opts = DatanodeOptions::default();
174 let toml_string = toml::to_string(&opts).unwrap();
175 let _parsed: DatanodeOptions = toml::from_str(&toml_string).unwrap();
176 }
177
178 #[test]
179 fn test_secstr() {
180 let toml_str = r#"
181 [storage]
182 type = "S3"
183 access_key_id = "access_key_id"
184 secret_access_key = "secret_access_key"
185 "#;
186 let opts: DatanodeOptions = toml::from_str(toml_str).unwrap();
187 match &opts.storage.store {
188 ObjectStoreConfig::S3(cfg) => {
189 assert_eq!(
190 "SecretBox<alloc::string::String>([REDACTED])".to_string(),
191 format!("{:?}", cfg.access_key_id)
192 );
193 assert_eq!("access_key_id", cfg.access_key_id.expose_secret());
194 }
195 _ => unreachable!(),
196 }
197 }
198 #[test]
199 fn test_skip_ssl_validation_config() {
200 let toml_str_true = r#"
202 [storage]
203 type = "S3"
204 [storage.http_client]
205 skip_ssl_validation = true
206 "#;
207 let opts: DatanodeOptions = toml::from_str(toml_str_true).unwrap();
208 match &opts.storage.store {
209 ObjectStoreConfig::S3(cfg) => {
210 assert!(cfg.http_client.skip_ssl_validation);
211 }
212 _ => panic!("Expected S3 config"),
213 }
214
215 let toml_str_false = r#"
217 [storage]
218 type = "S3"
219 [storage.http_client]
220 skip_ssl_validation = false
221 "#;
222 let opts: DatanodeOptions = toml::from_str(toml_str_false).unwrap();
223 match &opts.storage.store {
224 ObjectStoreConfig::S3(cfg) => {
225 assert!(!cfg.http_client.skip_ssl_validation);
226 }
227 _ => panic!("Expected S3 config"),
228 }
229 let toml_str_default = r#"
231 [storage]
232 type = "S3"
233 "#;
234 let opts: DatanodeOptions = toml::from_str(toml_str_default).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}