diff --git a/config/config.md b/config/config.md index d450609a926..e370bb157a0 100644 --- a/config/config.md +++ b/config/config.md @@ -174,7 +174,7 @@ | `storage.http_client.timeout` | String | `30s` | The total request timeout, applied from when the request starts connecting until the response body has finished.
Also considered a total deadline. | | `storage.http_client.pool_idle_timeout` | String | `90s` | The timeout for idle sockets being kept-alive. | | `storage.http_client.skip_ssl_validation` | Bool | `false` | To skip the ssl verification
**Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks. | -| `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines.
Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail. | +| `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines.
Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail.
The default configuration lists `mito` and `file`; `metric` is always registered, so naming
it only supplies its options rather than enabling the engine. | | `region_engine.mito` | -- | -- | The Mito engine options. | | `region_engine.mito.num_workers` | Integer | Auto | Number of region workers. Defaults to half of the available CPU cores, rounded up. | | `region_engine.mito.worker_channel_size` | Integer | `128` | Request channel size of each worker. | @@ -650,7 +650,7 @@ | `storage.http_client.timeout` | String | `30s` | The total request timeout, applied from when the request starts connecting until the response body has finished.
Also considered a total deadline. | | `storage.http_client.pool_idle_timeout` | String | `90s` | The timeout for idle sockets being kept-alive. | | `storage.http_client.skip_ssl_validation` | Bool | `false` | To skip the ssl verification
**Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks. | -| `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines.
Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail. | +| `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines.
Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail.
The default configuration lists `mito` and `file`; `metric` is always registered, so naming
it only supplies its options rather than enabling the engine. | | `region_engine.mito` | -- | -- | The Mito engine options. | | `region_engine.mito.num_workers` | Integer | Auto | Number of region workers. Defaults to half of the available CPU cores, rounded up. | | `region_engine.mito.worker_channel_size` | Integer | `128` | Request channel size of each worker. | diff --git a/config/datanode.example.toml b/config/datanode.example.toml index 4b8eb41242e..bfa3afc0bcf 100644 --- a/config/datanode.example.toml +++ b/config/datanode.example.toml @@ -509,6 +509,8 @@ skip_ssl_validation = false ## The region engine options. You can configure multiple region engines. ## Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail. +## The default configuration lists `mito` and `file`; `metric` is always registered, so naming +## it only supplies its options rather than enabling the engine. [[region_engine]] ## The Mito engine options. diff --git a/config/standalone.example.toml b/config/standalone.example.toml index f70277aa599..f73a3f8a325 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -729,6 +729,8 @@ skip_ssl_validation = false ## The region engine options. You can configure multiple region engines. ## Each engine type (mito, file, metric) may appear only once; duplicates cause startup to fail. +## The default configuration lists `mito` and `file`; `metric` is always registered, so naming +## it only supplies its options rather than enabling the engine. [[region_engine]] ## The Mito engine options. diff --git a/src/cmd/src/options.rs b/src/cmd/src/options.rs index e09b3c8f733..ed88431e4fd 100644 --- a/src/cmd/src/options.rs +++ b/src/cmd/src/options.rs @@ -65,6 +65,10 @@ where fn env_list_keys() -> Option<&'static [&'static str]> { T::env_list_keys() } + + fn env_tagged_seq_keys() -> Option<&'static [&'static str]> { + T::env_tagged_seq_keys() + } } /// A serde `deserialize_with` helper that parses the `plugins` field leniently. diff --git a/src/cmd/tests/load_config_test.rs b/src/cmd/tests/load_config_test.rs index 2781183c0c8..2a41c120b23 100644 --- a/src/cmd/tests/load_config_test.rs +++ b/src/cmd/tests/load_config_test.rs @@ -902,3 +902,216 @@ fn test_load_metric_config_with_removed_sparse_primary_key_encoding() { Duration::from_secs(30) ); } + +/// Returns the mito engine config of a loaded `region_engine` sequence. +fn find_mito(region_engine: &[RegionEngineConfig]) -> &MitoConfig { + region_engine + .iter() + .find_map(|engine| match engine { + RegionEngineConfig::Mito(config) => Some(config), + _ => None, + }) + .expect("the mito engine config") +} + +#[test] +fn test_load_region_engine_mito_field_from_env() { + let env_prefix = "REGION_ENGINE_MITO_FIELD_UT"; + let env_key = [ + env_prefix, + "REGION_ENGINE", + "MITO", + "GLOBAL_WRITE_BUFFER_REJECT_SIZE", + ] + .join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("4GB"), || { + let options = + GreptimeOptions::::load_layered_options(None, env_prefix).unwrap(); + + // The addressed field wins, the fields next to it keep their default value. + let mito = find_mito(&options.component.region_engine); + assert_eq!(mito.global_write_buffer_reject_size, ReadableSize::gb(4)); + assert_eq!( + mito.global_write_buffer_size, + MitoConfig::default().global_write_buffer_size + ); + assert_eq!(mito.index, MitoConfig::default().index); + + // The other entries of the default sequence are intact. + assert!( + options + .component + .region_engine + .iter() + .any(|engine| matches!(engine, RegionEngineConfig::File(_))), + "the file engine config should survive an env override of the mito engine" + ); + }); +} + +#[test] +fn test_load_region_engine_mito_field_from_env_standalone() { + let env_prefix = "REGION_ENGINE_MITO_FIELD_STANDALONE_UT"; + let env_key = [ + env_prefix, + "REGION_ENGINE", + "MITO", + "GLOBAL_WRITE_BUFFER_REJECT_SIZE", + ] + .join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("4GB"), || { + let options = + GreptimeOptions::::load_layered_options(None, env_prefix).unwrap(); + + let mito = find_mito(&options.component.region_engine); + assert_eq!(mito.global_write_buffer_reject_size, ReadableSize::gb(4)); + assert_eq!( + mito.global_write_buffer_size, + MitoConfig::default().global_write_buffer_size + ); + assert!( + options + .component + .region_engine + .iter() + .any(|engine| matches!(engine, RegionEngineConfig::File(_))), + "the file engine config should survive an env override of the mito engine" + ); + }); +} + +#[test] +fn test_load_region_engine_metric_entry_from_env() { + let env_prefix = "REGION_ENGINE_METRIC_FIELD_UT"; + let env_key = [ + env_prefix, + "REGION_ENGINE", + "METRIC", + "FLUSH_METADATA_REGION_INTERVAL", + ] + .join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("1m"), || { + let options = + GreptimeOptions::::load_layered_options(None, env_prefix).unwrap(); + + // `metric` is not in the default sequence, so it is appended instead of replacing + // the whole sequence. + assert_eq!( + options.component.region_engine.len(), + DatanodeOptions::default().region_engine.len() + 1 + ); + let metric = options + .component + .region_engine + .iter() + .find_map(|engine| match engine { + RegionEngineConfig::Metric(config) => Some(config), + _ => None, + }) + .expect("the metric engine config"); + assert_eq!( + metric.flush_metadata_region_interval, + Duration::from_secs(60) + ); + + // The engines of the default sequence keep their default options. + let mito = find_mito(&options.component.region_engine); + assert_eq!( + mito.global_write_buffer_reject_size, + MitoConfig::default().global_write_buffer_reject_size + ); + assert!( + options + .component + .region_engine + .iter() + .any(|engine| matches!(engine, RegionEngineConfig::File(_))) + ); + }); +} + +#[test] +fn test_load_region_engine_env_patch_is_replaced_by_config_file() { + let config = tempfile::NamedTempFile::new().unwrap(); + std::fs::write( + config.path(), + "[[region_engine]]\n[region_engine.metric]\nflush_metadata_region_interval = \"2m\"\n", + ) + .unwrap(); + + let env_prefix = "REGION_ENGINE_FILE_WINS_UT"; + let env_key = [ + env_prefix, + "REGION_ENGINE", + "MITO", + "GLOBAL_WRITE_BUFFER_REJECT_SIZE", + ] + .join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("4GB"), || { + let options = GreptimeOptions::::load_layered_options( + config.path().to_str(), + env_prefix, + ) + .unwrap(); + + // A sequence is a single value: the config file replaces the whole value produced by + // the environment, so the environment patch (which is merged onto the default + // entries) is dropped. That keeps the precedence file > env > default of every other + // option, and means a config file has to spell out every engine it wants. + assert_eq!( + options.component.region_engine, + vec![RegionEngineConfig::Metric(MetricEngineConfig { + flush_metadata_region_interval: Duration::from_secs(120), + })] + ); + }); +} + +#[test] +fn test_load_region_engine_env_patch_with_config_file_lacking_region_engine() { + // The usual deployment shape: a config file that sets other options, plus an environment + // variable patching one region engine. The file says nothing about `region_engine`, so + // there is nothing that could replace the patched sequence. + let config = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(config.path(), "node_id = 7\n").unwrap(); + + let env_prefix = "REGION_ENGINE_FILE_NO_ENGINE_UT"; + let env_key = [ + env_prefix, + "REGION_ENGINE", + "MITO", + "GLOBAL_WRITE_BUFFER_REJECT_SIZE", + ] + .join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("4GB"), || { + let options = GreptimeOptions::::load_layered_options( + config.path().to_str(), + env_prefix, + ) + .unwrap(); + + // The option of the file is applied... + assert_eq!(options.component.node_id, Some(7)); + // ...the environment patch applies on top of the default engine entries... + let mito = find_mito(&options.component.region_engine); + assert_eq!(mito.global_write_buffer_reject_size, ReadableSize::gb(4)); + assert_eq!( + mito.global_write_buffer_size, + MitoConfig::default().global_write_buffer_size + ); + // ...and the rest of the default sequence is still configured. + assert!( + options + .component + .region_engine + .iter() + .any(|engine| matches!(engine, RegionEngineConfig::File(_))), + "the file engine entry should survive" + ); + }); +} diff --git a/src/common/config/src/config.rs b/src/common/config/src/config.rs index 85ce3d206f6..8dfd195d0bc 100644 --- a/src/common/config/src/config.rs +++ b/src/common/config/src/config.rs @@ -17,7 +17,10 @@ use serde::Serialize; use serde::de::DeserializeOwned; use snafu::ResultExt; -use crate::error::{LoadLayeredConfigSnafu, Result, SerdeJsonSnafu, TomlFormatSnafu}; +use crate::error::{ + LoadLayeredConfigSnafu, Result, SerdeJsonSnafu, TaggedSeqNotTopLevelSnafu, TomlFormatSnafu, +}; +use crate::tagged_seq::TaggedSeqEnvSource; /// Separator for environment variables. For example, `DATANODE__STORAGE__MANIFEST__CHECKPOINT_MARGIN`. pub const ENV_VAR_SEP: &str = "__"; @@ -64,9 +67,27 @@ pub trait Configurable: Serialize + DeserializeOwned + Default + Sized { let default_config = File::from_str(&json_str, FileFormat::Json); // Add default values and environment variables as the sources of the configuration. - let mut layered_config = config::Config::builder() - .add_source(default_config) - .add_source(env_source); + let mut layered_config = config::Config::builder().add_source(default_config); + + // A sequence of tagged tables cannot be addressed by plain dotted env keys: the path + // walk of `config` drops the default sequence. Such keys are merged onto the default + // entries by tag name instead. + layered_config = match Self::env_tagged_seq_keys() { + Some(keys) if !keys.is_empty() => { + // The default entries are looked up by literal key, so a dotted key would find + // none of them. Reject it instead of silently dropping the defaults. + if let Some(key) = keys.iter().find(|key| key.contains('.')) { + return TaggedSeqNotTopLevelSnafu { + key: (*key).to_string(), + } + .fail(); + } + + let defaults = serde_json::to_value(&default_opts).context(SerdeJsonSnafu)?; + layered_config.add_source(TaggedSeqEnvSource::new(env_source, keys, &defaults)) + } + _ => layered_config.add_source(env_source), + }; // Add config file as the source of the configuration if it is specified. if let Some(config_file) = config_file { @@ -93,6 +114,22 @@ pub trait Configurable: Serialize + DeserializeOwned + Default + Sized { None } + /// Keys whose value is a sequence of single-key tagged tables (e.g. `region_engine`, + /// serialized from `Vec` with externally tagged variants). Environment variables + /// address them by tag name (`KEY____`) instead of numeric index, and the + /// overrides are merged onto the corresponding default entries. + /// + /// Keys must name a **top-level** field of the options struct: the default entries to + /// merge onto are looked up by literal key in the serialized defaults, so a dotted key is + /// rejected instead of silently leaving the defaults out of the merged sequence. + /// + /// A configuration file that configures such a key as a whole replaces the merged value, + /// so the environment variables for it are ineffective then and the file has to spell out + /// every entry it wants. + fn env_tagged_seq_keys() -> Option<&'static [&'static str]> { + None + } + /// Serialize the configuration to a TOML string. fn to_toml(&self) -> Result { toml::to_string(&self).context(TomlFormatSnafu) @@ -102,7 +139,9 @@ pub trait Configurable: Serialize + DeserializeOwned + Default + Sized { #[cfg(test)] mod tests { use std::io::Write; + use std::time::Duration; + use common_base::readable_size::ReadableSize; use common_telemetry::logging::LoggingOptions; use common_test_util::temp_dir::create_named_temp_file; use common_wal::config::DatanodeWalConfig; @@ -264,4 +303,358 @@ mod tests { }, ); } + + /// A sequence of single-key tagged tables, the shape of `Vec` with externally + /// tagged variants. + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] + #[serde(default)] + struct TestIndexConfig { + aux_path: String, + cache_percent: u8, + } + + impl Default for TestIndexConfig { + fn default() -> Self { + Self { + aux_path: "/default/aux".to_string(), + cache_percent: 30, + } + } + } + + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] + #[serde(default)] + struct TestMitoConfig { + global_write_buffer_size: ReadableSize, + global_write_buffer_reject_size: ReadableSize, + index: TestIndexConfig, + } + + impl Default for TestMitoConfig { + fn default() -> Self { + Self { + global_write_buffer_size: ReadableSize::mb(64), + global_write_buffer_reject_size: ReadableSize::gb(2), + index: TestIndexConfig::default(), + } + } + } + + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] + #[serde(default)] + struct TestFileConfig { + data_dir: String, + } + + impl Default for TestFileConfig { + fn default() -> Self { + Self { + data_dir: "/default/data".to_string(), + } + } + } + + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] + #[serde(default)] + struct TestMetricConfig { + #[serde(with = "humantime_serde")] + flush_metadata_region_interval: Duration, + } + + impl Default for TestMetricConfig { + fn default() -> Self { + Self { + flush_metadata_region_interval: Duration::from_secs(30), + } + } + } + + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] + #[serde(rename_all = "snake_case")] + enum TestRegionEngine { + Mito(TestMitoConfig), + File(TestFileConfig), + Metric(TestMetricConfig), + } + + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] + struct TestTaggedSeqConfig { + region_engine: Vec, + } + + impl Default for TestTaggedSeqConfig { + fn default() -> Self { + Self { + region_engine: vec![ + TestRegionEngine::Mito(TestMitoConfig::default()), + TestRegionEngine::File(TestFileConfig::default()), + ], + } + } + } + + impl Configurable for TestTaggedSeqConfig { + fn env_tagged_seq_keys() -> Option<&'static [&'static str]> { + Some(&["region_engine"]) + } + } + + /// A config whose tagged sequence key is not a top-level field. + #[derive(Debug, Serialize, Deserialize, Default)] + struct TestDottedKeyConfig { + region_engine: Vec, + } + + impl Configurable for TestDottedKeyConfig { + fn env_tagged_seq_keys() -> Option<&'static [&'static str]> { + Some(&["region_engine.mito"]) + } + } + + fn test_mito(config: &TestTaggedSeqConfig) -> &TestMitoConfig { + config + .region_engine + .iter() + .find_map(|engine| match engine { + TestRegionEngine::Mito(config) => Some(config), + _ => None, + }) + .expect("the mito entry") + } + + fn test_metric(config: &TestTaggedSeqConfig) -> &TestMetricConfig { + config + .region_engine + .iter() + .find_map(|engine| match engine { + TestRegionEngine::Metric(config) => Some(config), + _ => None, + }) + .expect("the metric entry") + } + + /// Loads with the given prefix and returns the message of the error, including the + /// source of the layered loading error. + fn load_error_text(env_prefix: &str) -> String { + let err = TestTaggedSeqConfig::load_layered_options(None, env_prefix).unwrap_err(); + let mut text = err.to_string(); + let mut source = std::error::Error::source(&err); + while let Some(error) = source { + text.push_str(&format!(": {error}")); + source = std::error::Error::source(error); + } + text + } + + #[test] + fn test_tagged_seq_env_overrides_a_field() { + let env_prefix = "TAGGED_SEQ_FIELD_UT"; + let env_key = [ + env_prefix, + "REGION_ENGINE", + "MITO", + "GLOBAL_WRITE_BUFFER_REJECT_SIZE", + ] + .join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("4GB"), || { + let opts = TestTaggedSeqConfig::load_layered_options(None, env_prefix).unwrap(); + let defaults = TestTaggedSeqConfig::default(); + + // The entries of the sequence that the environment does not mention are intact. + assert_eq!(opts.region_engine.len(), defaults.region_engine.len()); + assert_eq!(opts.region_engine[1], defaults.region_engine[1]); + + // The addressed field wins, the fields next to it keep their default value. + let mito = test_mito(&opts); + assert_eq!(mito.global_write_buffer_reject_size, ReadableSize::gb(4)); + assert_eq!( + mito.global_write_buffer_size, + TestMitoConfig::default().global_write_buffer_size + ); + assert_eq!(mito.index, TestIndexConfig::default()); + }); + } + + #[test] + fn test_tagged_seq_env_appends_an_entry() { + let env_prefix = "TAGGED_SEQ_APPEND_UT"; + let env_key = [ + env_prefix, + "REGION_ENGINE", + "METRIC", + "FLUSH_METADATA_REGION_INTERVAL", + ] + .join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("1m"), || { + let opts = TestTaggedSeqConfig::load_layered_options(None, env_prefix).unwrap(); + let defaults = TestTaggedSeqConfig::default(); + + // The entry is not in the default sequence, so it is appended. + assert_eq!(opts.region_engine.len(), defaults.region_engine.len() + 1); + assert_eq!(opts.region_engine[0], defaults.region_engine[0]); + assert_eq!(opts.region_engine[1], defaults.region_engine[1]); + assert_eq!( + test_metric(&opts).flush_metadata_region_interval, + Duration::from_secs(60) + ); + }); + } + + #[test] + fn test_tagged_seq_env_overrides_a_nested_field() { + let env_prefix = "TAGGED_SEQ_NESTED_UT"; + let env_key = [env_prefix, "REGION_ENGINE", "MITO", "INDEX", "AUX_PATH"].join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("/tmp/aux"), || { + let opts = TestTaggedSeqConfig::load_layered_options(None, env_prefix).unwrap(); + + let mito = test_mito(&opts); + assert_eq!(mito.index.aux_path, "/tmp/aux"); + assert_eq!( + mito.index.cache_percent, + TestIndexConfig::default().cache_percent + ); + assert_eq!( + mito.global_write_buffer_size, + TestMitoConfig::default().global_write_buffer_size + ); + }); + } + + #[test] + fn test_tagged_seq_env_rejects_whole_key() { + let env_prefix = "TAGGED_SEQ_NO_TAG_UT"; + let env_key = [env_prefix, "REGION_ENGINE"].join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("mito"), || { + let text = load_error_text(env_prefix); + assert!(text.contains("region_engine"), "{text}"); + assert!(text.contains("region_engine____"), "{text}"); + }); + } + + #[test] + fn test_tagged_seq_env_rejects_numeric_index() { + let env_prefix = "TAGGED_SEQ_NUMERIC_UT"; + let env_key = [ + env_prefix, + "REGION_ENGINE", + "0", + "MITO", + "GLOBAL_WRITE_BUFFER_SIZE", + ] + .join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("4GB"), || { + let text = load_error_text(env_prefix); + assert!( + text.contains("region_engine.0.mito.global_write_buffer_size"), + "{text}" + ); + assert!(text.contains("addressed by tag name"), "{text}"); + }); + } + + #[test] + fn test_tagged_seq_env_rejects_scalar_and_field_under_same_key() { + let env_prefix = "TAGGED_SEQ_NOT_A_TABLE_UT"; + let whole_key = [env_prefix, "REGION_ENGINE", "MITO", "INDEX"].join(ENV_VAR_SEP); + let below_key = + [env_prefix, "REGION_ENGINE", "MITO", "INDEX", "AUX_PATH"].join(ENV_VAR_SEP); + + temp_env::with_vars( + [ + (whole_key, Some("not-a-table")), + (below_key, Some("/tmp/aux")), + ], + || { + // One variable sets `index` as a whole, another sets a field below it. The + // conflict is reported as such instead of failing later inside serde with a + // type error that does not name the variables involved. + let text = load_error_text(env_prefix); + assert!( + text.contains("another environment variable sets 'region_engine.mito.index'"), + "{text}" + ); + assert!(text.contains("region_engine.mito.index.aux_path"), "{text}"); + }, + ); + } + + #[test] + fn test_tagged_seq_env_rejects_signed_and_float_indices() { + for index in ["-1", "+1", "1e3"] { + let env_prefix = "TAGGED_SEQ_INDEX_SHAPE_UT"; + let env_key = [ + env_prefix, + "REGION_ENGINE", + index, + "MITO", + "GLOBAL_WRITE_BUFFER_SIZE", + ] + .join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("4GB"), || { + let text = load_error_text(env_prefix); + assert!(text.contains("addressed by tag name"), "{index}: {text}"); + }); + } + } + + #[test] + fn test_tagged_seq_env_rejects_tag_without_field() { + let env_prefix = "TAGGED_SEQ_NO_FIELD_UT"; + let env_key = [env_prefix, "REGION_ENGINE", "MITO"].join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("mito"), || { + let text = load_error_text(env_prefix); + assert!(text.contains("region_engine.mito"), "{text}"); + assert!(text.contains("region_engine____"), "{text}"); + }); + } + + #[test] + fn test_tagged_seq_env_rejects_field_below_scalar_default() { + // The `data_dir` field of the file engine defaults to a string, so nothing can be set + // below it. + let env_prefix = "TAGGED_SEQ_SCALAR_DEFAULT_UT"; + let env_key = [env_prefix, "REGION_ENGINE", "FILE", "DATA_DIR", "SUB"].join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("/tmp/sub"), || { + let text = load_error_text(env_prefix); + assert!(text.contains("region_engine.file.data_dir"), "{text}"); + assert!(text.contains("default value is not a table"), "{text}"); + }); + } + + #[test] + fn test_tagged_seq_env_type_error_names_the_environment() { + let env_prefix = "TAGGED_SEQ_TYPE_ERROR_UT"; + let env_key = [ + env_prefix, + "REGION_ENGINE", + "MITO", + "INDEX", + "CACHE_PERCENT", + ] + .join(ENV_VAR_SEP); + + temp_env::with_var(env_key, Some("not-a-number"), || { + // The override replaces the default value as the environment collected it, so the + // type error still says where that value comes from. + let text = load_error_text(env_prefix); + assert!(text.contains("in the environment"), "{text}"); + }); + } + + #[test] + fn test_tagged_seq_key_must_be_top_level() { + let err = TestDottedKeyConfig::load_layered_options(None, "TAGGED_SEQ_DOTTED_KEY_UT") + .unwrap_err(); + let text = err.to_string(); + assert!(text.contains("region_engine.mito"), "{text}"); + assert!(text.contains("top-level"), "{text}"); + } } diff --git a/src/common/config/src/error.rs b/src/common/config/src/error.rs index 9e788d1d300..d7f81014e05 100644 --- a/src/common/config/src/error.rs +++ b/src/common/config/src/error.rs @@ -65,6 +65,84 @@ pub enum Error { #[snafu(implicit)] location: Location, }, + + #[snafu(display( + "Invalid environment variable key '{key}': '{target}' is a sequence that cannot be \ + overridden as a whole; address one of its entries with '{target}____'" + ))] + TaggedSeqMissingTag { + key: String, + target: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display( + "Invalid environment variable key '{key}': '{tag}' is a numeric index, but '{target}' \ + is addressed by tag name; use '{target}____' instead, e.g. \ + '{target}__MITO__'" + ))] + TaggedSeqNumericTag { + key: String, + tag: String, + target: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display( + "Invalid environment variable key '{key}': the entry '{tag}' of '{target}' needs a \ + field to override; use '{target}____' instead" + ))] + TaggedSeqMissingField { + key: String, + tag: String, + target: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display( + "Invalid environment variable key '{key}': another environment variable sets '{path}' \ + as a whole, so this key cannot set a field below it; set either the whole '{path}' or \ + its fields, not both" + ))] + TaggedSeqOverridesConflict { + key: String, + path: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display( + "'{path}' cannot be overridden from the environment: its default value is not a table, \ + so a field below it cannot be set" + ))] + TaggedSeqNotATable { + path: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display( + "'{key}' is registered as a sequence of tagged tables, but it is not a top-level field \ + of the options; the default entries to merge onto are looked up by literal key" + ))] + TaggedSeqNotTopLevel { + key: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display( + "The default value of '{target}' is not a sequence of single-key tables: {reason}" + ))] + TaggedSeqInvalidDefault { + target: String, + reason: String, + #[snafu(implicit)] + location: Location, + }, } impl ErrorExt for Error { @@ -73,7 +151,14 @@ impl ErrorExt for Error { Error::TomlFormat { .. } | Error::LoadLayeredConfig { .. } | Error::FileWatch { .. } - | Error::InvalidPath { .. } => StatusCode::InvalidArguments, + | Error::InvalidPath { .. } + | Error::TaggedSeqMissingTag { .. } + | Error::TaggedSeqNumericTag { .. } + | Error::TaggedSeqMissingField { .. } + | Error::TaggedSeqOverridesConflict { .. } + | Error::TaggedSeqNotATable { .. } + | Error::TaggedSeqNotTopLevel { .. } + | Error::TaggedSeqInvalidDefault { .. } => StatusCode::InvalidArguments, Error::SerdeJson { .. } => StatusCode::Unexpected, } } diff --git a/src/common/config/src/lib.rs b/src/common/config/src/lib.rs index eea3b1351d8..7dc962fe5b1 100644 --- a/src/common/config/src/lib.rs +++ b/src/common/config/src/lib.rs @@ -15,6 +15,7 @@ pub mod config; pub mod error; pub mod file_watcher; +mod tagged_seq; use std::time::Duration; diff --git a/src/common/config/src/tagged_seq.rs b/src/common/config/src/tagged_seq.rs new file mode 100644 index 00000000000..7d2faa0b9bd --- /dev/null +++ b/src/common/config/src/tagged_seq.rs @@ -0,0 +1,711 @@ +// 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. + +//! Environment overrides for configuration keys holding a sequence of tagged tables. +//! +//! Such a sequence is what a `Vec` of externally tagged enums serializes to, for +//! example the `region_engine` option: +//! +//! ```text +//! [{"mito": {...}}, {"file": {...}}] +//! ``` +//! +//! Handing the plain [`Environment`] source to `config` cannot work for these keys: +//! `config` writes every flat dotted key by walking down its path, and the walk resets +//! any intermediate value that is not a table. Walking `region_engine.mito.index.aux_path` +//! therefore destroys the default sequence before anything is set. +//! +//! This source instead addresses the entries by their tag name and merges the +//! overrides onto the matching default entries: +//! +//! ```text +//! GREPTIMEDB_STANDALONE__REGION_ENGINE__MITO__INDEX__AUX_PATH=... +//! ``` +//! +//! Entries that only the environment mentions are appended, and every field that is +//! not mentioned keeps its default value. A configuration file that configures the +//! whole sequence replaces the merged one, which makes these variables ineffective; +//! the file has to spell out every entry it wants. + +use config::{ConfigError, Environment, Map, Source, Value, ValueKind}; +use serde_json::Value as JsonValue; + +use crate::error::{ + Error, Result, TaggedSeqInvalidDefaultSnafu, TaggedSeqMissingFieldSnafu, + TaggedSeqMissingTagSnafu, TaggedSeqNotATableSnafu, TaggedSeqNumericTagSnafu, + TaggedSeqOverridesConflictSnafu, +}; + +/// A [`Source`] that merges environment overrides into sequences of tagged tables. +/// +/// See the [module documentation](self) for the addressed syntax. +#[derive(Debug)] +pub(crate) struct TaggedSeqEnvSource { + /// The wrapped environment source. Everything it collects is kept as is, apart + /// from the entries that address one of the `targets`. + inner: Environment, + /// The keys whose entries are addressed by tag name. + targets: Vec, +} + +/// One sequence of tagged tables that the environment may override. +#[derive(Debug, Clone)] +struct Target { + /// The dotted key of the sequence, e.g. `region_engine`. + key: String, + /// The default value of `key`, whose entries the overrides are merged into. + default: Option, +} + +impl TaggedSeqEnvSource { + /// Wraps `inner` so that the sequence at every key of `keys` accepts tag name + /// overrides from the environment, merged onto the value that the same key has + /// in `defaults`. + pub(crate) fn new( + inner: Environment, + keys: &'static [&'static str], + defaults: &JsonValue, + ) -> Self { + let targets = keys + .iter() + .map(|key| Target { + key: (*key).to_string(), + default: defaults.get(key).cloned(), + }) + .collect(); + + Self { inner, targets } + } +} + +impl Source for TaggedSeqEnvSource { + fn clone_into_box(&self) -> Box { + Box::new(Self { + inner: self.inner.clone(), + targets: self.targets.clone(), + }) + } + + fn collect(&self) -> std::result::Result, ConfigError> { + let mut collected = self.inner.collect()?; + + for target in &self.targets { + // Take the entries out first: `config` walks the collected map in an + // unspecified order, so a leftover `region_engine.mito.x` entry could be + // written after the sequence and destroy it again. + let entries = take_entries(&mut collected, &target.key); + let merged = merge(target, entries).map_err(|e| ConfigError::Message(e.to_string()))?; + collected.insert(target.key.clone(), merged); + } + + Ok(collected) + } +} + +/// Removes and returns the entries that address `key` itself or one of its children. +fn take_entries(collected: &mut Map, key: &str) -> Vec<(String, Value)> { + let child_prefix = format!("{key}."); + let mut keys = collected + .keys() + .filter(|existing| *existing == key || existing.starts_with(&child_prefix)) + .cloned() + .collect::>(); + // `Map` is unordered, sort so that a reported error is deterministic. + keys.sort(); + + keys.into_iter() + .filter_map(|key| collected.remove(&key).map(|value| (key, value))) + .collect() +} + +/// Rebuilds the sequence of `target` with the environment overrides `entries` applied. +fn merge(target: &Target, entries: Vec<(String, Value)>) -> Result { + let mut overrides = group_entries(target, entries)?; + + let mut merged = Vec::with_capacity(overrides.len() + 1); + for entry in default_entries(target)? { + let mut entry = json_to_value(entry); + let tag = single_tag(target, &entry)?; + + if let Some(value) = overrides.remove(&tag) { + apply_override(target, &tag, &mut entry, value)?; + } + + merged.push(entry); + } + + // Entries that the default sequence does not have are appended. + let mut appended = overrides.into_iter().collect::>(); + appended.sort_by(|a, b| a.0.cmp(&b.0)); + for (tag, value) in appended { + let mut table = Map::new(); + table.insert(tag, value); + merged.push(Value::new(None, ValueKind::Table(table))); + } + + Ok(Value::new(None, ValueKind::Array(merged))) +} + +/// Merges the override of the entry `tag` into `entry`, whose only key is `tag`. +fn apply_override(target: &Target, tag: &str, entry: &mut Value, value: Value) -> Result<()> { + let ValueKind::Table(body) = &mut entry.kind else { + return invalid_default(target, "expected an array of tables"); + }; + + let Some(field) = body.get_mut(tag) else { + return invalid_default(target, "expected each entry to hold exactly one tag"); + }; + + // Both sides are tables, this only fails on a field whose default value cannot hold + // the sub-table that the environment sets. + deep_merge(field, value).map_err(|path| not_a_table(&entry_path(target, tag, &path))) +} + +/// Returns the default entries of `target`, each one a single-key table `{tag: table}`. +fn default_entries(target: &Target) -> Result<&[JsonValue]> { + match &target.default { + None | Some(JsonValue::Null) => Ok(&[]), + Some(JsonValue::Array(entries)) => Ok(entries), + Some(_) => invalid_default(target, "expected an array of tables"), + } +} + +/// Returns the tag of `entry`, which must map one tag to a table. +fn single_tag(target: &Target, entry: &Value) -> Result { + let ValueKind::Table(table) = &entry.kind else { + return invalid_default(target, "expected an array of tables"); + }; + + let mut keys = table.keys(); + let (Some(tag), None) = (keys.next(), keys.next()) else { + return invalid_default(target, "expected each entry to hold exactly one tag"); + }; + if !matches!(&table[tag].kind, ValueKind::Table(_)) { + return invalid_default(target, "expected each entry to map its tag to a table"); + } + + Ok(tag.clone()) +} + +/// Groups the overrides by tag, nesting every field path into a table. +fn group_entries(target: &Target, entries: Vec<(String, Value)>) -> Result> { + // Every entry but `key` itself is addressed as `.....`. + let offset = target.key.len() + 1; + let mut groups: Map = Map::new(); + + for (key, value) in entries { + if key == target.key { + return TaggedSeqMissingTagSnafu { + key, + target: target.key.clone(), + } + .fail(); + } + + let rest = key[offset..].to_string(); + let Some((tag, field)) = rest.split_once('.') else { + return TaggedSeqMissingFieldSnafu { + key, + tag: rest, + target: target.key.clone(), + } + .fail(); + }; + + if is_index_like(tag) { + return TaggedSeqNumericTagSnafu { + key, + tag: tag.to_string(), + target: target.key.clone(), + } + .fail(); + } + + let field = nest(field, value); + match groups.get_mut(tag) { + Some(group) => deep_merge(group, field) + .map_err(|path| overrides_conflict(&key, &entry_path(target, tag, &path)))?, + None => { + groups.insert(tag.to_string(), field); + } + } + } + + Ok(groups) +} + +/// Whether `tag` looks like a position in the sequence rather than an engine name. +/// +/// `config` also accepts negative and floating point looking path segments (`[-1]`, +/// `[1e3]`), so anything that starts like a number is reported as a misplaced index; +/// otherwise those variables would only fail later with the generic `unknown variant` +/// error from serde. +fn is_index_like(tag: &str) -> bool { + let mut chars = tag.chars(); + match chars.next() { + Some(first) if first.is_ascii_digit() => true, + Some('-') | Some('+') => { + matches!(chars.next(), Some(next) if next.is_ascii_digit() || next == '.') + } + _ => false, + } +} + +/// Builds a table whose key path is the dotted `path`, e.g. `a.b` and `v` give +/// `{a: {b: v}}`. +fn nest(path: &str, value: Value) -> Value { + let mut table = Map::new(); + match path.split_once('.') { + Some((head, tail)) => { + table.insert(head.to_string(), nest(tail, value)); + } + None => { + table.insert(path.to_string(), value); + } + } + + Value::new(None, ValueKind::Table(table)) +} + +/// Merges `incoming` into `base`, recursively for tables. Returns the dotted path of +/// the field that cannot be merged into, when `base` holds a non-table value where +/// `incoming` needs a table. +fn deep_merge(base: &mut Value, incoming: Value) -> std::result::Result<(), String> { + merge_at(base, incoming, "") +} + +fn merge_at(base: &mut Value, incoming: Value, path: &str) -> std::result::Result<(), String> { + match incoming.kind { + ValueKind::Table(incoming) => { + let ValueKind::Table(base) = &mut base.kind else { + return Err(path.to_string()); + }; + for (field, value) in incoming { + let field_path = if path.is_empty() { + field.clone() + } else { + format!("{path}.{field}") + }; + // A non-table override replaces whatever the key already holds. It is inserted + // as is, so the value keeps the origin it was collected with. + if matches!(value.kind, ValueKind::Table(_)) { + match base.get_mut(&field) { + Some(existing) => merge_at(existing, value, &field_path)?, + None => { + base.insert(field, value); + } + } + } else { + base.insert(field, value); + } + } + Ok(()) + } + kind => { + // A non-table override replaces whatever the key already holds. + base.kind = kind; + Ok(()) + } + } +} + +fn invalid_default(target: &Target, reason: &str) -> Result { + TaggedSeqInvalidDefaultSnafu { + target: target.key.clone(), + reason: reason.to_string(), + } + .fail() +} + +/// Builds the dotted path of `path` below the entry `tag` of `target`. +fn entry_path(target: &Target, tag: &str, path: &str) -> String { + if path.is_empty() { + format!("{}.{}", target.key, tag) + } else { + format!("{}.{}.{}", target.key, tag, path) + } +} + +/// Error for a field whose default value is not a table, so nothing can be set below it. +fn not_a_table(path: &str) -> Error { + TaggedSeqNotATableSnafu { + path: path.to_string(), + } + .build() +} + +/// Error for a key that sets a field below a path another key sets as a whole. +fn overrides_conflict(key: &str, path: &str) -> Error { + TaggedSeqOverridesConflictSnafu { + key: key.to_string(), + path: path.to_string(), + } + .build() +} + +/// Converts a default value parsed from the JSON serialization of the options. +fn json_to_value(json: &JsonValue) -> Value { + let kind = match json { + JsonValue::Null => ValueKind::Nil, + JsonValue::Bool(value) => ValueKind::Boolean(*value), + JsonValue::Number(value) => match (value.as_i64(), value.as_u64()) { + (Some(value), _) => ValueKind::I64(value), + (None, Some(value)) => ValueKind::U64(value), + (None, None) => ValueKind::Float(value.as_f64().unwrap_or_default()), + }, + JsonValue::String(value) => ValueKind::String(value.clone()), + JsonValue::Array(values) => ValueKind::Array(values.iter().map(json_to_value).collect()), + JsonValue::Object(values) => ValueKind::Table( + values + .iter() + .map(|(key, value)| (key.clone(), json_to_value(value))) + .collect(), + ), + }; + + Value::new(None, kind) +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + const KEY: &str = "region_engine"; + const KEYS: &[&str] = &[KEY]; + + /// The default sequence used by the tests: two exported entries, `mito` before `file`, + /// with one nested table to override. + fn defaults() -> JsonValue { + json!({ + "region_engine": [ + { + "mito": { + "global_write_buffer_size": "64MB", + "index": { + "aux_path": "/default/aux", + "cache_percent": 30 + } + } + }, + {"file": {"data_dir": "/default/data"}} + ], + "data_home": "/default/data_home" + }) + } + + /// A fixed set of variables, so the tests never touch the process environment and can + /// run in parallel. + fn env(vars: &[(&str, &str)]) -> Environment { + let mut vars_map = Map::new(); + for (key, value) in vars { + vars_map.insert((*key).to_string(), (*value).to_string()); + } + + Environment::default() + .separator("__") + .try_parsing(true) + .source(Some(vars_map)) + } + + fn collect_with(vars: &[(&str, &str)], defaults: &JsonValue) -> Map { + TaggedSeqEnvSource::new(env(vars), KEYS, defaults) + .collect() + .unwrap() + } + + fn collect(vars: &[(&str, &str)]) -> Map { + collect_with(vars, &defaults()) + } + + fn to_json(collected: Map) -> JsonValue { + let mut map = serde_json::Map::new(); + for (key, value) in collected { + map.insert(key, value.try_deserialize().unwrap()); + } + + JsonValue::Object(map) + } + + fn collect_json(vars: &[(&str, &str)]) -> JsonValue { + to_json(collect(vars)) + } + + fn collect_json_with(vars: &[(&str, &str)], defaults: &JsonValue) -> JsonValue { + to_json(collect_with(vars, defaults)) + } + + fn collect_error_with(vars: &[(&str, &str)], defaults: &JsonValue) -> String { + match TaggedSeqEnvSource::new(env(vars), KEYS, defaults).collect() { + Ok(_) => panic!("expected an error, got {vars:?}"), + Err(ConfigError::Message(message)) => message, + Err(other) => panic!("expected a message error, got: {other}"), + } + } + + fn collect_error(vars: &[(&str, &str)]) -> String { + collect_error_with(vars, &defaults()) + } + + #[test] + fn test_collect_overrides_a_scalar_field() { + let collected = collect_json(&[("REGION_ENGINE__MITO__GLOBAL_WRITE_BUFFER_SIZE", "128MB")]); + + let engines = collected["region_engine"].as_array().unwrap(); + assert_eq!(engines.len(), 2); + assert_eq!(engines[0]["mito"]["global_write_buffer_size"], "128MB"); + // The fields next to the override keep their default value. + assert_eq!(engines[0]["mito"]["index"]["aux_path"], "/default/aux"); + assert_eq!(engines[1]["file"]["data_dir"], "/default/data"); + } + + #[test] + fn test_collect_overrides_a_nested_field() { + let collected = collect_json(&[("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux")]); + + assert_eq!( + collected["region_engine"][0]["mito"]["index"]["aux_path"], + "/tmp/aux" + ); + assert_eq!( + collected["region_engine"][0]["mito"]["index"]["cache_percent"], + 30 + ); + assert_eq!( + collected["region_engine"][0]["mito"]["global_write_buffer_size"], + "64MB" + ); + } + + #[test] + fn test_collect_merges_several_variables_of_one_entry() { + let collected = collect_json(&[ + ("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux"), + ("REGION_ENGINE__MITO__INDEX__CACHE_PERCENT", "50"), + ]); + + assert_eq!( + collected["region_engine"][0]["mito"]["index"]["aux_path"], + "/tmp/aux" + ); + assert_eq!( + collected["region_engine"][0]["mito"]["index"]["cache_percent"], + 50 + ); + } + + #[test] + fn test_collect_appends_unknown_tags_in_tag_order() { + let collected = collect_json(&[ + ( + "REGION_ENGINE__METRIC__FLUSH_METADATA_REGION_INTERVAL", + "1m", + ), + ("REGION_ENGINE__AAA__DATA_DIR", "/tmp/aaa"), + ]); + + let engines = collected["region_engine"].as_array().unwrap(); + assert_eq!(engines.len(), 4); + // The default entries stay first and keep their order. + assert_eq!(engines[0]["mito"]["global_write_buffer_size"], "64MB"); + assert_eq!(engines[1]["file"]["data_dir"], "/default/data"); + // Entries that only the environment mentions are appended, sorted by tag. + assert_eq!(engines[2], json!({"aaa": {"data_dir": "/tmp/aaa"}})); + assert_eq!( + engines[3], + json!({"metric": {"flush_metadata_region_interval": "1m"}}) + ); + } + + #[test] + fn test_collect_keeps_the_default_sequence_without_overrides() { + let collected = collect_json(&[("DATA_HOME", "/srv/data")]); + + assert_eq!(collected["region_engine"], defaults()["region_engine"]); + assert_eq!(collected["data_home"], "/srv/data"); + } + + #[test] + fn test_collect_removes_the_flat_keys_of_the_sequence() { + let collected = collect(&[ + ("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux"), + ("REGION_ENGINE_EXTRA__MITO__AUX_PATH", "/tmp/other"), + ]); + + // A leftover `region_engine.x` key would be written after the sequence and destroy + // it again, so all of them must be consumed by the merge. + assert!( + !collected + .keys() + .any(|key| key.starts_with("region_engine.")), + "flat keys left in the collected map: {:?}", + collected.keys().collect::>() + ); + // A key that only shares the prefix is not part of the sequence. + assert!(collected.contains_key("region_engine_extra.mito.aux_path")); + } + + #[test] + fn test_collect_without_default_entries() { + let collected = collect_json_with( + &[ + ( + "REGION_ENGINE__METRIC__FLUSH_METADATA_REGION_INTERVAL", + "1m", + ), + ("REGION_ENGINE__FILE__DATA_DIR", "/tmp/data"), + ], + &json!({}), + ); + + assert_eq!( + collected["region_engine"], + json!([ + {"file": {"data_dir": "/tmp/data"}}, + {"metric": {"flush_metadata_region_interval": "1m"}} + ]) + ); + } + + #[test] + fn test_collect_with_null_default_entries() { + let collected = collect_json_with( + &[("REGION_ENGINE__FILE__DATA_DIR", "/tmp/data")], + &json!({"region_engine": null}), + ); + + assert_eq!( + collected["region_engine"], + json!([{"file": {"data_dir": "/tmp/data"}}]) + ); + } + + #[test] + fn test_collect_rejects_setting_the_whole_sequence() { + let message = collect_error(&[("REGION_ENGINE", "mito")]); + + assert!( + message.contains("cannot be overridden as a whole"), + "{message}" + ); + assert!( + message.contains("region_engine____"), + "{message}" + ); + } + + #[test] + fn test_collect_rejects_a_numeric_index() { + let message = collect_error(&[("REGION_ENGINE__0__MITO__INDEX__AUX_PATH", "/tmp/aux")]); + + assert!(message.contains("numeric index"), "{message}"); + assert!(message.contains("addressed by tag name"), "{message}"); + } + + #[test] + fn test_collect_rejects_a_tag_without_field() { + let message = collect_error(&[("REGION_ENGINE__MITO", "mito")]); + + assert!(message.contains("needs a field"), "{message}"); + } + + #[test] + fn test_collect_rejects_conflicting_overrides() { + let message = collect_error(&[ + ("REGION_ENGINE__MITO__INDEX", "not-a-table"), + ("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux"), + ]); + + assert!( + message.contains("another environment variable sets 'region_engine.mito.index'"), + "{message}" + ); + } + + #[test] + fn test_collect_rejects_a_field_below_a_scalar_default() { + let message = collect_error(&[("REGION_ENGINE__FILE__DATA_DIR__SUB", "/tmp/sub")]); + + assert!(message.contains("region_engine.file.data_dir"), "{message}"); + assert!( + message.contains("default value is not a table"), + "{message}" + ); + } + + #[test] + fn test_collect_rejects_a_scalar_default_sequence() { + let message = collect_error_with( + &[("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux")], + &json!({"region_engine": "not-an-array"}), + ); + + assert!( + message.contains("not a sequence of single-key tables"), + "{message}" + ); + } + + #[test] + fn test_collect_rejects_a_default_entry_with_two_tags() { + let message = collect_error_with( + &[("REGION_ENGINE__MITO__INDEX__AUX_PATH", "/tmp/aux")], + &json!({"region_engine": [{"mito": {}, "file": {}}]}), + ); + + assert!(message.contains("exactly one tag"), "{message}"); + } + + #[test] + fn test_take_entries_keeps_unrelated_keys() { + let mut collected = Map::new(); + for key in [ + "region_engine", + "region_engine.mito.index", + "region_engine_extra.mito", + "other", + ] { + collected.insert( + (*key).to_string(), + Value::new(None, ValueKind::String("v".to_string())), + ); + } + + let entries = take_entries(&mut collected, "region_engine"); + + assert_eq!( + entries + .iter() + .map(|(key, _)| key.as_str()) + .collect::>(), + vec!["region_engine", "region_engine.mito.index"] + ); + assert!(!collected.contains_key("region_engine")); + assert!(!collected.contains_key("region_engine.mito.index")); + assert!(collected.contains_key("region_engine_extra.mito")); + assert!(collected.contains_key("other")); + } + + #[test] + fn test_is_index_like() { + for tag in ["0", "12", "-1", "+1", "1e3", "-1.5"] { + assert!(is_index_like(tag), "{tag} should look like an index"); + } + + for tag in ["mito", "file", "metric", "", "m1", "-", "+"] { + assert!(!is_index_like(tag), "{tag} should not look like an index"); + } + } +} diff --git a/src/datanode/src/config.rs b/src/datanode/src/config.rs index e627d8dc245..34a41ce2801 100644 --- a/src/datanode/src/config.rs +++ b/src/datanode/src/config.rs @@ -173,6 +173,10 @@ impl Configurable for DatanodeOptions { "wal.broker_endpoints", ]) } + + fn env_tagged_seq_keys() -> Option<&'static [&'static str]> { + Some(&["region_engine"]) + } } #[allow(clippy::large_enum_variant)] diff --git a/src/standalone/src/options.rs b/src/standalone/src/options.rs index 657961e9298..bbbad14b93d 100644 --- a/src/standalone/src/options.rs +++ b/src/standalone/src/options.rs @@ -137,6 +137,10 @@ impl Configurable for StandaloneOptions { "pending_rows_batcher.logical_table.protocols", ]) } + + fn env_tagged_seq_keys() -> Option<&'static [&'static str]> { + Some(&["region_engine"]) + } } /// The [`StandaloneOptions`] is only defined in `standalone` crate,