standalone/
options.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::time::Duration;
16
17use common_base::readable_size::ReadableSize;
18use common_base::secrets::SecretString;
19use common_config::{Configurable, KvBackendConfig};
20use common_event_recorder::EventRecorderOptions;
21use common_memory_manager::OnExhaustedPolicy;
22use common_options::memory::MemoryOptions;
23use common_telemetry::logging::{LoggingOptions, SlowQueryOptions, TracingOptions};
24use common_wal::config::DatanodeWalConfig;
25use datanode::config::{DatanodeOptions, ProcedureConfig, RegionEngineConfig, StorageConfig};
26use file_engine::config::EngineConfig as FileEngineConfig;
27use flow::FlowConfig;
28use frontend::frontend::FrontendOptions;
29use frontend::service_config::{
30    InfluxdbOptions, JaegerOptions, MysqlOptions, OpentsdbOptions, OtlpOptions,
31    PendingRowsBatcherOptions, PostgresOptions, PromStoreOptions, WriteRetryOptions,
32};
33use mito2::config::MitoConfig;
34use pipeline::PipelineOptions;
35use query::options::QueryOptions;
36use serde::{Deserialize, Serialize};
37use servers::grpc::GrpcOptions;
38use servers::http::HttpOptions;
39
40#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
41#[serde(default)]
42pub struct StandaloneOptions {
43    pub enable_telemetry: bool,
44    pub default_timezone: Option<String>,
45    pub default_column_prefix: Option<String>,
46    /// Server-side global switch for auto table creation on write.
47    /// Upper bound: when `false`, missing tables are never auto-created even if a
48    /// request sets the `auto_create_table` hint to `true`. Default: `true`.
49    pub auto_create_table: bool,
50    /// Maximum total memory for all concurrent write request bodies and messages (HTTP, gRPC, Flight).
51    /// Set to 0 to disable the limit. Default: "0" (unlimited)
52    pub max_in_flight_write_bytes: ReadableSize,
53    /// Policy when write bytes quota is exhausted.
54    /// Options: "wait" (default, 10s), "wait(<duration>)", "fail"
55    pub write_bytes_exhausted_policy: OnExhaustedPolicy,
56    pub http: HttpOptions,
57    pub grpc: GrpcOptions,
58    pub mysql: MysqlOptions,
59    pub postgres: PostgresOptions,
60    pub opentsdb: OpentsdbOptions,
61    pub influxdb: InfluxdbOptions,
62    /// Ordinary-table batching with independent logical-table controls.
63    pub pending_rows_batcher: PendingRowsBatcherOptions,
64    pub jaeger: JaegerOptions,
65    pub otlp: OtlpOptions,
66    pub prom_store: PromStoreOptions,
67    pub wal: DatanodeWalConfig,
68    pub storage: StorageConfig,
69    pub metadata_store: KvBackendConfig,
70    pub procedure: ProcedureConfig,
71    pub flow: FlowConfig,
72    pub logging: LoggingOptions,
73    /// The user provider for authentication.
74    ///
75    /// The inline `static_user_provider:cmd:` form carries plaintext
76    /// passwords, so the value is held as a [`SecretString`] and serialized
77    /// with those passwords redacted.
78    #[serde(serialize_with = "auth::serialize_user_provider_option")]
79    pub user_provider: Option<SecretString>,
80    /// Options for different store engines.
81    pub region_engine: Vec<RegionEngineConfig>,
82    pub tracing: TracingOptions,
83    pub init_regions_in_background: bool,
84    pub init_regions_parallelism: usize,
85    /// Maximum concurrent region-read streams. Zero means unlimited.
86    /// A single query can consume multiple permits.
87    pub max_concurrent_queries: usize,
88    /// Timeout to acquire a permit when `max_concurrent_queries` is reached.
89    /// Only effective when the limiter is enabled.
90    #[serde(with = "humantime_serde")]
91    pub concurrent_query_limiter_timeout: Duration,
92    pub slow_query: SlowQueryOptions,
93    pub query: QueryOptions,
94    pub memory: MemoryOptions,
95    /// The pipeline options.
96    pub pipeline: PipelineOptions,
97    /// The event recorder options.
98    pub event_recorder: EventRecorderOptions,
99    /// Environment variable keys to read and report in heartbeat messages.
100    pub heartbeat_env_vars: Vec<String>,
101}
102
103impl Default for StandaloneOptions {
104    fn default() -> Self {
105        Self {
106            enable_telemetry: true,
107            default_timezone: None,
108            default_column_prefix: None,
109            auto_create_table: true,
110            max_in_flight_write_bytes: ReadableSize(0),
111            write_bytes_exhausted_policy: OnExhaustedPolicy::default(),
112            http: HttpOptions::default(),
113            grpc: GrpcOptions::default(),
114            mysql: MysqlOptions::default(),
115            postgres: PostgresOptions::default(),
116            opentsdb: OpentsdbOptions::default(),
117            influxdb: InfluxdbOptions::default(),
118            pending_rows_batcher: PendingRowsBatcherOptions::default(),
119            jaeger: JaegerOptions::default(),
120            otlp: OtlpOptions::default(),
121            prom_store: PromStoreOptions::default(),
122            wal: DatanodeWalConfig::default(),
123            storage: StorageConfig::default(),
124            metadata_store: KvBackendConfig::default(),
125            procedure: ProcedureConfig::default(),
126            flow: FlowConfig::default(),
127            logging: LoggingOptions::default(),
128            user_provider: None,
129            region_engine: vec![
130                RegionEngineConfig::Mito(MitoConfig::default()),
131                RegionEngineConfig::File(FileEngineConfig::default()),
132            ],
133            tracing: TracingOptions::default(),
134            init_regions_in_background: false,
135            init_regions_parallelism: 16,
136            max_concurrent_queries: 0,
137            concurrent_query_limiter_timeout: Duration::from_millis(100),
138            slow_query: SlowQueryOptions::default(),
139            query: QueryOptions::default(),
140            memory: MemoryOptions::default(),
141            pipeline: PipelineOptions::default(),
142            event_recorder: EventRecorderOptions::default(),
143            heartbeat_env_vars: vec![],
144        }
145    }
146}
147
148impl Configurable for StandaloneOptions {
149    fn env_list_keys() -> Option<&'static [&'static str]> {
150        Some(&[
151            "heartbeat_env_vars",
152            "wal.broker_endpoints",
153            "event_recorder.event_types",
154            "pending_rows_batcher.protocols",
155            "pending_rows_batcher.logical_table.protocols",
156        ])
157    }
158
159    fn env_tagged_seq_keys() -> Option<&'static [&'static str]> {
160        Some(&["region_engine"])
161    }
162}
163
164/// The [`StandaloneOptions`] is only defined in `standalone` crate,
165/// we don't want to make `frontend` depends on it, so impl [`Into`]
166/// rather than [`From`].
167#[allow(clippy::from_over_into)]
168impl Into<FrontendOptions> for StandaloneOptions {
169    fn into(self) -> FrontendOptions {
170        self.frontend_options()
171    }
172}
173
174impl StandaloneOptions {
175    pub fn frontend_options(&self) -> FrontendOptions {
176        let cloned_opts = self.clone();
177        FrontendOptions {
178            default_timezone: cloned_opts.default_timezone,
179            auto_create_table: cloned_opts.auto_create_table,
180            max_in_flight_write_bytes: cloned_opts.max_in_flight_write_bytes,
181            write_bytes_exhausted_policy: cloned_opts.write_bytes_exhausted_policy,
182            http: cloned_opts.http,
183            grpc: cloned_opts.grpc,
184            mysql: cloned_opts.mysql,
185            postgres: cloned_opts.postgres,
186            opentsdb: cloned_opts.opentsdb,
187            influxdb: cloned_opts.influxdb,
188            pending_rows_batcher: cloned_opts.pending_rows_batcher,
189            jaeger: cloned_opts.jaeger,
190            otlp: cloned_opts.otlp,
191            prom_store: cloned_opts.prom_store,
192            meta_client: None,
193            // Local region calls consume requests directly and do not use remote write retries.
194            write_retry: WriteRetryOptions {
195                max_retries: 0,
196                ..Default::default()
197            },
198            logging: cloned_opts.logging,
199            user_provider: cloned_opts.user_provider,
200            query: cloned_opts.query,
201            slow_query: cloned_opts.slow_query,
202            pipeline: cloned_opts.pipeline,
203            event_recorder: cloned_opts.event_recorder,
204            heartbeat_env_vars: cloned_opts.heartbeat_env_vars.clone(),
205            ..Default::default()
206        }
207    }
208
209    pub fn datanode_options(&self) -> DatanodeOptions {
210        let cloned_opts = self.clone();
211        DatanodeOptions {
212            node_id: Some(0),
213            enable_telemetry: cloned_opts.enable_telemetry,
214            wal: cloned_opts.wal,
215            storage: cloned_opts.storage,
216            region_engine: cloned_opts.region_engine,
217            grpc: cloned_opts.grpc,
218            init_regions_in_background: cloned_opts.init_regions_in_background,
219            init_regions_parallelism: cloned_opts.init_regions_parallelism,
220            max_concurrent_queries: cloned_opts.max_concurrent_queries,
221            concurrent_query_limiter_timeout: cloned_opts.concurrent_query_limiter_timeout,
222            query: cloned_opts.query,
223            heartbeat_env_vars: cloned_opts.heartbeat_env_vars,
224            ..Default::default()
225        }
226    }
227
228    /// Sanitize the `StandaloneOptions` to ensure the config is valid.
229    pub fn sanitize(&mut self) {
230        if self.storage.is_object_storage() {
231            self.storage
232                .store
233                .cache_config_mut()
234                .unwrap()
235                .sanitize(&self.storage.data_home);
236        }
237    }
238}
239
240#[cfg(test)]
241mod tests {
242    use std::sync::Arc;
243
244    use common_config::Configurable;
245    use common_event_recorder::EventTypeFilter;
246    use servers::batcher::BatchingProtocol;
247
248    use crate::options::*;
249
250    #[test]
251    fn test_user_provider_redacts_cmd_passwords_in_config() {
252        let options = StandaloneOptions {
253            user_provider: Some("static_user_provider:cmd:alice=secret".to_string().into()),
254            ..Default::default()
255        };
256
257        let serialized = options.to_toml().unwrap();
258        assert!(serialized.contains("user_provider = \"[REDACTED]\""));
259        assert!(!serialized.contains("secret"));
260    }
261
262    #[test]
263    fn test_concurrent_query_limiter_options() {
264        let defaults: StandaloneOptions = toml::from_str("").unwrap();
265        let datanode_defaults = DatanodeOptions::default();
266        assert_eq!(defaults.max_concurrent_queries, 0);
267        assert_eq!(
268            defaults.concurrent_query_limiter_timeout,
269            Duration::from_millis(100)
270        );
271        assert_eq!(
272            defaults.max_concurrent_queries,
273            datanode_defaults.max_concurrent_queries
274        );
275        assert_eq!(
276            defaults.concurrent_query_limiter_timeout,
277            datanode_defaults.concurrent_query_limiter_timeout
278        );
279
280        let options: StandaloneOptions = toml::from_str(
281            r#"max_concurrent_queries = 8
282concurrent_query_limiter_timeout = "250ms""#,
283        )
284        .unwrap();
285        let restored: StandaloneOptions =
286            toml::from_str(&toml::to_string(&options).unwrap()).unwrap();
287        for options in [options, restored] {
288            assert_eq!(options.max_concurrent_queries, 8);
289            assert_eq!(
290                options.concurrent_query_limiter_timeout,
291                Duration::from_millis(250)
292            );
293            let datanode = options.datanode_options();
294            assert_eq!(datanode.max_concurrent_queries, 8);
295            assert_eq!(
296                datanode.concurrent_query_limiter_timeout,
297                Duration::from_millis(250)
298            );
299        }
300        assert!(
301            toml::from_str::<StandaloneOptions>("concurrent_query_limiter_timeout = 'invalid'")
302                .is_err()
303        );
304    }
305
306    #[test]
307    fn test_concurrent_query_limiter_options_from_env() {
308        temp_env::with_vars(
309            [
310                ("STANDALONE_LIMITER_TEST__MAX_CONCURRENT_QUERIES", Some("4")),
311                (
312                    "STANDALONE_LIMITER_TEST__CONCURRENT_QUERY_LIMITER_TIMEOUT",
313                    Some("2s"),
314                ),
315            ],
316            || {
317                let options =
318                    StandaloneOptions::load_layered_options(None, "STANDALONE_LIMITER_TEST")
319                        .unwrap();
320                let datanode = options.datanode_options();
321                assert_eq!(datanode.max_concurrent_queries, 4);
322                assert_eq!(
323                    datanode.concurrent_query_limiter_timeout,
324                    Duration::from_secs(2)
325                );
326            },
327        );
328    }
329
330    #[test]
331    fn test_pipeline_config_forwarding() {
332        let defaults: StandaloneOptions = toml::from_str("").unwrap();
333        assert!(!defaults.pipeline.allow_vrl_host_access);
334        assert!(!defaults.frontend_options().pipeline.allow_vrl_host_access);
335        assert!(!defaults.pipeline.ignore_errors);
336        assert!(!defaults.frontend_options().pipeline.ignore_errors);
337        let options: StandaloneOptions =
338            toml::from_str("[pipeline]\nallow_vrl_host_access = true\nignore_errors = true")
339                .unwrap();
340        assert!(options.frontend_options().pipeline.allow_vrl_host_access);
341        assert!(options.frontend_options().pipeline.ignore_errors);
342        let serialized = toml::to_string(&options).unwrap();
343        let restored: StandaloneOptions = toml::from_str(&serialized).unwrap();
344        assert_eq!(options.pipeline, restored.pipeline);
345    }
346
347    #[test]
348    fn test_frontend_write_retries_disabled() {
349        assert!(FrontendOptions::default().write_retry.max_retries > 0);
350        assert_eq!(
351            StandaloneOptions::default()
352                .frontend_options()
353                .write_retry
354                .max_retries,
355            0
356        );
357    }
358
359    #[test]
360    fn test_logical_batcher_config_forwarding() {
361        let opts: StandaloneOptions = toml::from_str("[pending_rows_batcher]\nprotocols = ['influxdb']\n[pending_rows_batcher.logical_table]\nprotocols = ['otlp', 'prom']\npending_rows_flush_interval = '10ms'").unwrap();
362        let frontend = opts.frontend_options();
363        assert_eq!(frontend.pending_rows_batcher, opts.pending_rows_batcher);
364        assert_eq!(
365            frontend.pending_rows_batcher.logical_table,
366            opts.pending_rows_batcher.logical_table
367        );
368        let restored: StandaloneOptions = toml::from_str(&toml::to_string(&opts).unwrap()).unwrap();
369        assert_eq!(
370            restored.pending_rows_batcher.logical_table,
371            opts.pending_rows_batcher.logical_table
372        );
373        assert!(
374            toml::from_str::<StandaloneOptions>(
375                "[pending_rows_batcher.logical_table]\nprotocols = ['logs']"
376            )
377            .is_err()
378        );
379    }
380
381    #[test]
382    fn test_logical_batcher_protocols_from_env() {
383        temp_env::with_vars(
384            [
385                (
386                    "STANDALONE_LOGICAL_TEST__PENDING_ROWS_BATCHER__PROTOCOLS",
387                    Some("otlp,influxdb"),
388                ),
389                (
390                    "STANDALONE_LOGICAL_TEST__PENDING_ROWS_BATCHER__LOGICAL_TABLE__PROTOCOLS",
391                    Some("prom,otlp"),
392                ),
393            ],
394            || {
395                let options =
396                    StandaloneOptions::load_layered_options(None, "STANDALONE_LOGICAL_TEST")
397                        .unwrap();
398                assert_eq!(options.pending_rows_batcher.table.protocols.len(), 2);
399                assert_eq!(
400                    options
401                        .pending_rows_batcher
402                        .logical_table
403                        .unwrap()
404                        .protocols,
405                    vec![BatchingProtocol::Prom, BatchingProtocol::Otlp]
406                );
407            },
408        );
409    }
410
411    #[test]
412    fn test_batcher_protocols_from_env() {
413        temp_env::with_vars(
414            [(
415                "STANDALONE_BATCHER_TEST__PENDING_ROWS_BATCHER__PROTOCOLS",
416                Some("influxdb,http_sql"),
417            )],
418            || {
419                let options =
420                    StandaloneOptions::load_layered_options(None, "STANDALONE_BATCHER_TEST")
421                        .unwrap();
422                assert_eq!(
423                    options.pending_rows_batcher.table.protocols,
424                    vec![BatchingProtocol::Influxdb, BatchingProtocol::HttpSql]
425                );
426            },
427        );
428    }
429
430    #[test]
431    fn test_protocol_pending_rows_batcher_config() {
432        let defaults: StandaloneOptions = toml::from_str("").unwrap();
433        assert!(
434            !defaults
435                .pending_rows_batcher
436                .table
437                .pending_rows_batching_enabled()
438        );
439        let options: StandaloneOptions = toml::from_str(
440            r#"
441[pending_rows_batcher]
442protocols = ["influxdb", "http_sql"]
443pending_rows_flush_interval = "5ms"
444max_batch_rows = 25
445flow_notification_queue_capacity = 17
446"#,
447        )
448        .unwrap();
449        assert_eq!(options.pending_rows_batcher.table.max_batch_rows, 25);
450        assert_eq!(options.pending_rows_batcher.table.protocols.len(), 2);
451        assert_eq!(
452            options
453                .pending_rows_batcher
454                .table
455                .flow_notification_queue_capacity
456                .get(),
457            17
458        );
459        assert!(
460            options
461                .pending_rows_batcher
462                .table
463                .pending_rows_batching_enabled()
464        );
465        let serialized = toml::to_string(&options).unwrap();
466        let parsed: StandaloneOptions = toml::from_str(&serialized).unwrap();
467        assert_eq!(options.influxdb, parsed.influxdb);
468        assert_eq!(options.opentsdb, parsed.opentsdb);
469        assert_eq!(options.pending_rows_batcher, parsed.pending_rows_batcher);
470        let frontend = options.frontend_options();
471        assert_eq!(options.influxdb, frontend.influxdb);
472        assert_eq!(options.opentsdb, frontend.opentsdb);
473        assert_eq!(options.pending_rows_batcher, frontend.pending_rows_batcher);
474    }
475
476    #[test]
477    fn test_event_recorder_event_types_preserve_filter_semantics() {
478        let all: StandaloneOptions = toml::from_str("").unwrap();
479        let none: StandaloneOptions = toml::from_str("[event_recorder]\nevent_types = []").unwrap();
480        let selected: StandaloneOptions =
481            toml::from_str("[event_recorder]\nevent_types = ['create_database']").unwrap();
482
483        assert!(all.event_recorder.event_types.allows("future_event"));
484        assert_eq!(
485            none.event_recorder.event_types.as_ref(),
486            &EventTypeFilter::Only(Default::default())
487        );
488        assert!(
489            selected
490                .event_recorder
491                .event_types
492                .allows("create_database")
493        );
494        assert!(!selected.event_recorder.event_types.allows("drop_database"));
495
496        let frontend_options = selected.frontend_options();
497        assert!(Arc::ptr_eq(
498            &selected.event_recorder.event_types,
499            &frontend_options.event_recorder.event_types,
500        ));
501    }
502
503    #[test]
504    fn test_query_options_propagated_to_components() {
505        let mut options = StandaloneOptions::default();
506        options.query.parallelism = 4;
507
508        assert_eq!(options.frontend_options().query.parallelism, 4);
509        assert_eq!(options.datanode_options().query.parallelism, 4);
510    }
511}
standalone/options.rs

standalone/
options.rs
