greptimedb/src/standalone/src/options.rs

// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::time::Duration;

use common_base::readable_size::ReadableSize;
use common_config::{Configurable, KvBackendConfig};
use common_event_recorder::EventRecorderOptions;
use common_memory_manager::OnExhaustedPolicy;
use common_options::memory::MemoryOptions;
use common_telemetry::logging::{LoggingOptions, SlowQueryOptions, TracingOptions};
use common_wal::config::DatanodeWalConfig;
use datanode::config::{DatanodeOptions, ProcedureConfig, RegionEngineConfig, StorageConfig};
use file_engine::config::EngineConfig as FileEngineConfig;
use flow::FlowConfig;
use frontend::frontend::FrontendOptions;
use frontend::service_config::{
    InfluxdbOptions, JaegerOptions, MysqlOptions, OpentsdbOptions, OtlpOptions,
    PendingRowsBatcherOptions, PostgresOptions, PromStoreOptions, WriteRetryOptions,
};
use mito2::config::MitoConfig;
use pipeline::PipelineOptions;
use query::options::QueryOptions;
use serde::{Deserialize, Serialize};
use servers::grpc::GrpcOptions;
use servers::http::HttpOptions;

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
#[serde(default)]
pub struct StandaloneOptions {
    pub enable_telemetry: bool,
    pub default_timezone: Option<String>,
    pub default_column_prefix: Option<String>,
    /// Server-side global switch for auto table creation on write.
    /// Upper bound: when `false`, missing tables are never auto-created even if a
    /// request sets the `auto_create_table` hint to `true`. Default: `true`.
    pub auto_create_table: bool,
    /// Enables experimental Parquet database exports using shared Metric scans.
    pub experimental_metric_export: bool,
    /// Maximum total memory for all concurrent write request bodies and messages (HTTP, gRPC, Flight).
    /// Set to 0 to disable the limit. Default: "0" (unlimited)
    pub max_in_flight_write_bytes: ReadableSize,
    /// Policy when write bytes quota is exhausted.
    /// Options: "wait" (default, 10s), "wait(<duration>)", "fail"
    pub write_bytes_exhausted_policy: OnExhaustedPolicy,
    pub http: HttpOptions,
    pub grpc: GrpcOptions,
    pub mysql: MysqlOptions,
    pub postgres: PostgresOptions,
    pub opentsdb: OpentsdbOptions,
    pub influxdb: InfluxdbOptions,
    /// Ordinary-table batching with independent logical-table controls.
    pub pending_rows_batcher: PendingRowsBatcherOptions,
    pub jaeger: JaegerOptions,
    pub otlp: OtlpOptions,
    pub prom_store: PromStoreOptions,
    pub wal: DatanodeWalConfig,
    pub storage: StorageConfig,
    pub metadata_store: KvBackendConfig,
    pub procedure: ProcedureConfig,
    pub flow: FlowConfig,
    pub logging: LoggingOptions,
    pub user_provider: Option<String>,
    /// Options for different store engines.
    pub region_engine: Vec<RegionEngineConfig>,
    pub tracing: TracingOptions,
    pub init_regions_in_background: bool,
    pub init_regions_parallelism: usize,
    /// Maximum concurrent region-read streams. Zero means unlimited.
    /// A single query can consume multiple permits.
    pub max_concurrent_queries: usize,
    /// Timeout to acquire a permit when `max_concurrent_queries` is reached.
    /// Only effective when the limiter is enabled.
    #[serde(with = "humantime_serde")]
    pub concurrent_query_limiter_timeout: Duration,
    pub slow_query: SlowQueryOptions,
    pub query: QueryOptions,
    pub memory: MemoryOptions,
    /// The pipeline options.
    pub pipeline: PipelineOptions,
    /// The event recorder options.
    pub event_recorder: EventRecorderOptions,
    /// Environment variable keys to read and report in heartbeat messages.
    pub heartbeat_env_vars: Vec<String>,
}

impl Default for StandaloneOptions {
    fn default() -> Self {
        Self {
            enable_telemetry: true,
            default_timezone: None,
            default_column_prefix: None,
            auto_create_table: true,
            experimental_metric_export: false,
            max_in_flight_write_bytes: ReadableSize(0),
            write_bytes_exhausted_policy: OnExhaustedPolicy::default(),
            http: HttpOptions::default(),
            grpc: GrpcOptions::default(),
            mysql: MysqlOptions::default(),
            postgres: PostgresOptions::default(),
            opentsdb: OpentsdbOptions::default(),
            influxdb: InfluxdbOptions::default(),
            pending_rows_batcher: PendingRowsBatcherOptions::default(),
            jaeger: JaegerOptions::default(),
            otlp: OtlpOptions::default(),
            prom_store: PromStoreOptions::default(),
            wal: DatanodeWalConfig::default(),
            storage: StorageConfig::default(),
            metadata_store: KvBackendConfig::default(),
            procedure: ProcedureConfig::default(),
            flow: FlowConfig::default(),
            logging: LoggingOptions::default(),
            user_provider: None,
            region_engine: vec![
                RegionEngineConfig::Mito(MitoConfig::default()),
                RegionEngineConfig::File(FileEngineConfig::default()),
            ],
            tracing: TracingOptions::default(),
            init_regions_in_background: false,
            init_regions_parallelism: 16,
            max_concurrent_queries: 0,
            concurrent_query_limiter_timeout: Duration::from_millis(100),
            slow_query: SlowQueryOptions::default(),
            query: QueryOptions::default(),
            memory: MemoryOptions::default(),
            pipeline: PipelineOptions::default(),
            event_recorder: EventRecorderOptions::default(),
            heartbeat_env_vars: vec![],
        }
    }
}

impl Configurable for StandaloneOptions {
    fn env_list_keys() -> Option<&'static [&'static str]> {
        Some(&[
            "heartbeat_env_vars",
            "wal.broker_endpoints",
            "event_recorder.event_types",
            "pending_rows_batcher.protocols",
            "pending_rows_batcher.logical_table.protocols",
        ])
    }
}

/// The [`StandaloneOptions`] is only defined in `standalone` crate,
/// we don't want to make `frontend` depends on it, so impl [`Into`]
/// rather than [`From`].
#[allow(clippy::from_over_into)]
impl Into<FrontendOptions> for StandaloneOptions {
    fn into(self) -> FrontendOptions {
        self.frontend_options()
    }
}

impl StandaloneOptions {
    pub fn frontend_options(&self) -> FrontendOptions {
        let cloned_opts = self.clone();
        FrontendOptions {
            default_timezone: cloned_opts.default_timezone,
            auto_create_table: cloned_opts.auto_create_table,
            experimental_metric_export: cloned_opts.experimental_metric_export,
            max_in_flight_write_bytes: cloned_opts.max_in_flight_write_bytes,
            write_bytes_exhausted_policy: cloned_opts.write_bytes_exhausted_policy,
            http: cloned_opts.http,
            grpc: cloned_opts.grpc,
            mysql: cloned_opts.mysql,
            postgres: cloned_opts.postgres,
            opentsdb: cloned_opts.opentsdb,
            influxdb: cloned_opts.influxdb,
            pending_rows_batcher: cloned_opts.pending_rows_batcher,
            jaeger: cloned_opts.jaeger,
            otlp: cloned_opts.otlp,
            prom_store: cloned_opts.prom_store,
            meta_client: None,
            // Local region calls consume requests directly and do not use remote write retries.
            write_retry: WriteRetryOptions {
                max_retries: 0,
                ..Default::default()
            },
            logging: cloned_opts.logging,
            user_provider: cloned_opts.user_provider,
            query: cloned_opts.query,
            slow_query: cloned_opts.slow_query,
            pipeline: cloned_opts.pipeline,
            event_recorder: cloned_opts.event_recorder,
            heartbeat_env_vars: cloned_opts.heartbeat_env_vars.clone(),
            ..Default::default()
        }
    }

    pub fn datanode_options(&self) -> DatanodeOptions {
        let cloned_opts = self.clone();
        DatanodeOptions {
            node_id: Some(0),
            enable_telemetry: cloned_opts.enable_telemetry,
            wal: cloned_opts.wal,
            storage: cloned_opts.storage,
            region_engine: cloned_opts.region_engine,
            grpc: cloned_opts.grpc,
            init_regions_in_background: cloned_opts.init_regions_in_background,
            init_regions_parallelism: cloned_opts.init_regions_parallelism,
            max_concurrent_queries: cloned_opts.max_concurrent_queries,
            concurrent_query_limiter_timeout: cloned_opts.concurrent_query_limiter_timeout,
            query: cloned_opts.query,
            heartbeat_env_vars: cloned_opts.heartbeat_env_vars,
            ..Default::default()
        }
    }

    /// Sanitize the `StandaloneOptions` to ensure the config is valid.
    pub fn sanitize(&mut self) {
        if self.storage.is_object_storage() {
            self.storage
                .store
                .cache_config_mut()
                .unwrap()
                .sanitize(&self.storage.data_home);
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use common_event_recorder::EventTypeFilter;
    use servers::batcher::BatchingProtocol;

    use crate::options::*;

    #[test]
    fn test_frontend_write_retries_disabled() {
        assert!(FrontendOptions::default().write_retry.max_retries > 0);
        assert_eq!(
            StandaloneOptions::default()
                .frontend_options()
                .write_retry
                .max_retries,
            0
        );
    }

    #[test]
    fn test_concurrent_query_limiter_options() {
        let defaults: StandaloneOptions = toml::from_str("").unwrap();
        let datanode_defaults = DatanodeOptions::default();
        assert_eq!(defaults.max_concurrent_queries, 0);
        assert_eq!(
            defaults.concurrent_query_limiter_timeout,
            Duration::from_millis(100)
        );
        assert_eq!(
            defaults.max_concurrent_queries,
            datanode_defaults.max_concurrent_queries
        );
        assert_eq!(
            defaults.concurrent_query_limiter_timeout,
            datanode_defaults.concurrent_query_limiter_timeout
        );

        let options: StandaloneOptions = toml::from_str(
            r#"max_concurrent_queries = 8
concurrent_query_limiter_timeout = "250ms""#,
        )
        .unwrap();
        let restored: StandaloneOptions =
            toml::from_str(&toml::to_string(&options).unwrap()).unwrap();
        for options in [options, restored] {
            assert_eq!(options.max_concurrent_queries, 8);
            assert_eq!(
                options.concurrent_query_limiter_timeout,
                Duration::from_millis(250)
            );
            let datanode = options.datanode_options();
            assert_eq!(datanode.max_concurrent_queries, 8);
            assert_eq!(
                datanode.concurrent_query_limiter_timeout,
                Duration::from_millis(250)
            );
        }
        assert!(
            toml::from_str::<StandaloneOptions>("concurrent_query_limiter_timeout = 'invalid'")
                .is_err()
        );
    }

    #[test]
    fn test_concurrent_query_limiter_options_from_env() {
        temp_env::with_vars(
            [
                ("STANDALONE_LIMITER_TEST__MAX_CONCURRENT_QUERIES", Some("4")),
                (
                    "STANDALONE_LIMITER_TEST__CONCURRENT_QUERY_LIMITER_TIMEOUT",
                    Some("2s"),
                ),
            ],
            || {
                let options =
                    StandaloneOptions::load_layered_options(None, "STANDALONE_LIMITER_TEST")
                        .unwrap();
                let datanode = options.datanode_options();
                assert_eq!(datanode.max_concurrent_queries, 4);
                assert_eq!(
                    datanode.concurrent_query_limiter_timeout,
                    Duration::from_secs(2)
                );
            },
        );
    }

    #[test]
    fn test_pipeline_host_access_config_forwarding() {
        let defaults: StandaloneOptions = toml::from_str("").unwrap();
        assert!(!defaults.pipeline.allow_vrl_host_access);
        assert!(!defaults.frontend_options().pipeline.allow_vrl_host_access);
        let options: StandaloneOptions =
            toml::from_str("[pipeline]\nallow_vrl_host_access = true").unwrap();
        assert!(options.frontend_options().pipeline.allow_vrl_host_access);
        let serialized = toml::to_string(&options).unwrap();
        let restored: StandaloneOptions = toml::from_str(&serialized).unwrap();
        assert_eq!(options.pipeline, restored.pipeline);
    }

    #[test]
    fn test_logical_batcher_config_forwarding() {
        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();
        let frontend = opts.frontend_options();
        assert_eq!(frontend.pending_rows_batcher, opts.pending_rows_batcher);
        assert_eq!(
            frontend.pending_rows_batcher.logical_table,
            opts.pending_rows_batcher.logical_table
        );
        let restored: StandaloneOptions = toml::from_str(&toml::to_string(&opts).unwrap()).unwrap();
        assert_eq!(
            restored.pending_rows_batcher.logical_table,
            opts.pending_rows_batcher.logical_table
        );
        assert!(
            toml::from_str::<StandaloneOptions>(
                "[pending_rows_batcher.logical_table]\nprotocols = ['logs']"
            )
            .is_err()
        );
    }

    #[test]
    fn test_logical_batcher_protocols_from_env() {
        temp_env::with_vars(
            [
                (
                    "STANDALONE_LOGICAL_TEST__PENDING_ROWS_BATCHER__PROTOCOLS",
                    Some("otlp,influxdb"),
                ),
                (
                    "STANDALONE_LOGICAL_TEST__PENDING_ROWS_BATCHER__LOGICAL_TABLE__PROTOCOLS",
                    Some("prom,otlp"),
                ),
            ],
            || {
                let options =
                    StandaloneOptions::load_layered_options(None, "STANDALONE_LOGICAL_TEST")
                        .unwrap();
                assert_eq!(options.pending_rows_batcher.table.protocols.len(), 2);
                assert_eq!(
                    options
                        .pending_rows_batcher
                        .logical_table
                        .unwrap()
                        .protocols,
                    vec![BatchingProtocol::Prom, BatchingProtocol::Otlp]
                );
            },
        );
    }

    #[test]
    fn test_batcher_protocols_from_env() {
        temp_env::with_vars(
            [(
                "STANDALONE_BATCHER_TEST__PENDING_ROWS_BATCHER__PROTOCOLS",
                Some("influxdb,http_sql"),
            )],
            || {
                let options =
                    StandaloneOptions::load_layered_options(None, "STANDALONE_BATCHER_TEST")
                        .unwrap();
                assert_eq!(
                    options.pending_rows_batcher.table.protocols,
                    vec![BatchingProtocol::Influxdb, BatchingProtocol::HttpSql]
                );
            },
        );
    }

    #[test]
    fn test_protocol_pending_rows_batcher_config() {
        let defaults: StandaloneOptions = toml::from_str("").unwrap();
        assert!(
            !defaults
                .pending_rows_batcher
                .table
                .pending_rows_batching_enabled()
        );
        let options: StandaloneOptions = toml::from_str(
            r#"
[pending_rows_batcher]
protocols = ["influxdb", "http_sql"]
pending_rows_flush_interval = "5ms"
max_batch_rows = 25
flow_notification_queue_capacity = 17
"#,
        )
        .unwrap();
        assert_eq!(options.pending_rows_batcher.table.max_batch_rows, 25);
        assert_eq!(options.pending_rows_batcher.table.protocols.len(), 2);
        assert_eq!(
            options
                .pending_rows_batcher
                .table
                .flow_notification_queue_capacity
                .get(),
            17
        );
        assert!(
            options
                .pending_rows_batcher
                .table
                .pending_rows_batching_enabled()
        );
        let serialized = toml::to_string(&options).unwrap();
        let parsed: StandaloneOptions = toml::from_str(&serialized).unwrap();
        assert_eq!(options.influxdb, parsed.influxdb);
        assert_eq!(options.opentsdb, parsed.opentsdb);
        assert_eq!(options.pending_rows_batcher, parsed.pending_rows_batcher);
        let frontend = options.frontend_options();
        assert_eq!(options.influxdb, frontend.influxdb);
        assert_eq!(options.opentsdb, frontend.opentsdb);
        assert_eq!(options.pending_rows_batcher, frontend.pending_rows_batcher);
    }

    #[test]
    fn test_event_recorder_event_types_preserve_filter_semantics() {
        let all: StandaloneOptions = toml::from_str("").unwrap();
        let none: StandaloneOptions = toml::from_str("[event_recorder]\nevent_types = []").unwrap();
        let selected: StandaloneOptions =
            toml::from_str("[event_recorder]\nevent_types = ['create_database']").unwrap();

        assert!(all.event_recorder.event_types.allows("future_event"));
        assert_eq!(
            none.event_recorder.event_types.as_ref(),
            &EventTypeFilter::Only(Default::default())
        );
        assert!(
            selected
                .event_recorder
                .event_types
                .allows("create_database")
        );
        assert!(!selected.event_recorder.event_types.allows("drop_database"));

        let frontend_options = selected.frontend_options();
        assert!(Arc::ptr_eq(
            &selected.event_recorder.event_types,
            &frontend_options.event_recorder.event_types,
        ));
    }

    #[test]
    fn test_query_options_propagated_to_components() {
        let mut options = StandaloneOptions::default();
        assert!(!options.frontend_options().experimental_metric_export);
        options.experimental_metric_export = true;
        assert!(options.frontend_options().experimental_metric_export);
        options.query.parallelism = 4;

        assert_eq!(options.frontend_options().query.parallelism, 4);
        assert_eq!(options.datanode_options().query.parallelism, 4);
    }
}