diff --git a/src/cmd/src/datanode/builder.rs b/src/cmd/src/datanode/builder.rs index 7e4b83f09c..afabcb85f1 100644 --- a/src/cmd/src/datanode/builder.rs +++ b/src/cmd/src/datanode/builder.rs @@ -73,6 +73,7 @@ impl InstanceBuilder { None, ); + crate::options::flush_dropped_plugin_warnings(); log_versions(verbose_version(), short_version(), APP_NAME); maybe_activate_heap_profile(&dn_opts.memory); create_resource_limit_metrics(APP_NAME); diff --git a/src/cmd/src/flownode.rs b/src/cmd/src/flownode.rs index 126e1a4069..ee071875c3 100644 --- a/src/cmd/src/flownode.rs +++ b/src/cmd/src/flownode.rs @@ -258,6 +258,7 @@ impl StartCommand { None, ); + crate::options::flush_dropped_plugin_warnings(); log_versions(verbose_version(), short_version(), APP_NAME); maybe_activate_heap_profile(&opts.component.memory); create_resource_limit_metrics(APP_NAME); diff --git a/src/cmd/src/frontend.rs b/src/cmd/src/frontend.rs index e36b45b169..cbbd797eed 100644 --- a/src/cmd/src/frontend.rs +++ b/src/cmd/src/frontend.rs @@ -343,6 +343,7 @@ impl StartCommand { Some(&opts.component.slow_query), ); + crate::options::flush_dropped_plugin_warnings(); log_versions(verbose_version(), short_version(), APP_NAME); maybe_activate_heap_profile(&opts.component.memory); create_resource_limit_metrics(APP_NAME); diff --git a/src/cmd/src/metasrv.rs b/src/cmd/src/metasrv.rs index 28867df253..791efeb6e8 100644 --- a/src/cmd/src/metasrv.rs +++ b/src/cmd/src/metasrv.rs @@ -328,6 +328,7 @@ impl StartCommand { None, ); + crate::options::flush_dropped_plugin_warnings(); log_versions(verbose_version(), short_version(), APP_NAME); maybe_activate_heap_profile(&opts.component.memory); create_resource_limit_metrics(APP_NAME); diff --git a/src/cmd/src/options.rs b/src/cmd/src/options.rs index 0f72c980b2..e09b3c8f73 100644 --- a/src/cmd/src/options.rs +++ b/src/cmd/src/options.rs @@ -14,9 +14,12 @@ use clap::Parser; use common_config::Configurable; +use common_options::plugin_options::PluginOptionsDeserializer; use common_runtime::global::RuntimeOptions; use plugins::PluginOptions; +use plugins::options::PluginOptionsDeserializerImpl; use serde::{Deserialize, Serialize}; +use serde_json::Value; #[derive(Parser, Default, Debug, Clone)] pub struct GlobalOptions { @@ -42,6 +45,12 @@ pub struct GreptimeOptions { /// The runtime options. pub runtime: RuntimeOptions, /// The plugin options. + /// + /// Deserialized leniently via [`deserialize_plugin_options`]: plugin + /// options not recognized by the current build are dropped (with a deferred + /// warning) instead of aborting startup, while malformed payloads for known + /// variants still error. + #[serde(default, deserialize_with = "deserialize_plugin_options")] pub plugins: Vec, /// The options of each component (like Datanode or Standalone) of GreptimeDB. @@ -57,3 +66,40 @@ where T::env_list_keys() } } + +/// A serde `deserialize_with` helper that parses the `plugins` field leniently. +/// +/// It delegates to the plugin crate's deserializer (`PluginOptionsDeserializerImpl`), +/// which knows the recognized variants: unknown options are dropped while +/// malformed known options still error. Keeping the variant-aware logic in the +/// `plugins` crate avoids duplicating it and avoids introducing a new symbol on +/// the replaceable `plugins` crate. +fn deserialize_plugin_options<'de, D>(deserializer: D) -> Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + let values = Vec::::deserialize(deserializer)?; + if values.is_empty() { + return Ok(vec![]); + } + let payload = serde_json::to_string(&values).map_err(serde::de::Error::custom)?; + PluginOptionsDeserializerImpl + .deserialize(&payload) + .map_err(serde::de::Error::custom) +} + +/// Emits the plugin options that were dropped during config loading (because +/// they were not recognized by the current build) as warnings. +/// +/// Config loading runs before the global tracing subscriber is installed, so +/// the warnings are buffered and must be flushed once logging is initialized. +/// Each server's `build` method calls this right after initializing logging. +pub(crate) fn flush_dropped_plugin_warnings() { + let tags = common_options::plugin_options::take_dropped_plugin_warnings(); + for tag in tags { + common_telemetry::warn!( + "Ignoring unrecognized plugin option `{tag}` in the config; \ + it likely belongs to a plugin that is not compiled into this build." + ); + } +} diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs index 7a28366eb8..61823320cf 100644 --- a/src/cmd/src/standalone.rs +++ b/src/cmd/src/standalone.rs @@ -350,6 +350,7 @@ impl StartCommand { Some(&opts.component.slow_query), ); + crate::options::flush_dropped_plugin_warnings(); log_versions(verbose_version(), short_version(), APP_NAME); maybe_activate_heap_profile(&opts.component.memory); create_resource_limit_metrics(APP_NAME); @@ -1271,4 +1272,88 @@ mod tests { "test_data_home" ); } + + #[test] + fn test_load_options_ignores_unknown_plugin_options() { + // Plugin options that are not recognized by the current build (for example, + // an enterprise plugin option seen by an open-source build) must not abort + // startup. They should be dropped with a warning instead. + let mut file = create_named_temp_file(); + write!( + file, + r#" +[[plugins]] +SomeUnknownPlugin = {{ feature = "foo", count = 5 }} + +[[plugins]] +AnotherUnknownPlugin = {{}} +"# + ) + .unwrap(); + + let opts = GreptimeOptions::::load_layered_options( + Some(file.path().to_str().unwrap()), + "GREPTIMEDB_STANDALONE_UT", + ) + .expect( + "loading a config with unrecognized plugin options should succeed, \ + ignoring the unknown ones", + ); + // Unknown plugin options are dropped; the recognized list is empty in the + // open-source build. + assert!(opts.plugins.is_empty()); + } + + #[test] + fn test_load_options_errors_on_malformed_known_plugin_option() { + // A *known* variant with a malformed payload must NOT be silently + // dropped; config loading must fail so a misconfigured plugin is not + // disabled without notice. Here `Dummy` (a unit variant) is given a map + // payload, which is invalid. + let mut file = create_named_temp_file(); + write!( + file, + r#" +[[plugins]] +Dummy = {{ unexpected = "payload" }} +"# + ) + .unwrap(); + + let result = GreptimeOptions::::load_layered_options( + Some(file.path().to_str().unwrap()), + "GREPTIMEDB_STANDALONE_UT", + ); + assert!( + result.is_err(), + "a malformed payload for a known plugin variant must fail config loading" + ); + } + + #[test] + fn test_load_options_ignores_multi_key_unknown_plugin_entry() { + // A single plugin table carrying several *unknown* keys must not abort + // startup with serde's "expected map with a single key" error; it is + // dropped like any other unrecognized plugin option. + let mut file = create_named_temp_file(); + write!( + file, + r#" +[[plugins]] +FirstUnknownPlugin = {{ a = 1 }} +SecondUnknownPlugin = {{ b = 2 }} +"# + ) + .unwrap(); + + let opts = GreptimeOptions::::load_layered_options( + Some(file.path().to_str().unwrap()), + "GREPTIMEDB_STANDALONE_UT", + ) + .expect( + "loading a config with a multi-key unknown plugin entry should succeed, \ + ignoring the unknown ones", + ); + assert!(opts.plugins.is_empty()); + } } diff --git a/src/common/options/src/plugin_options.rs b/src/common/options/src/plugin_options.rs index fe5efe6ab7..6b03681540 100644 --- a/src/common/options/src/plugin_options.rs +++ b/src/common/options/src/plugin_options.rs @@ -12,9 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use serde::de::DeserializeOwned; +use serde_json::Value; /// A trait for serializing Metasrv config to a JSON string. /// So it can be used in the metasrv's crate instead of depending on the plugins' crate. @@ -34,3 +35,272 @@ pub trait PluginOptionsDeserializer: Send + Sync { /// so we add a flag to the plugins to indicate that the plugins are running in the standalone mode. #[derive(Clone, Copy, Debug)] pub struct StandaloneFlag; + +/// Buffer of plugin option tags that were dropped during config loading because +/// they are not recognized by the current build. +/// +/// Config loading typically happens *before* the global tracing subscriber is +/// installed, so a `warn!` emitted at that point is silently lost. Instead we +/// buffer the dropped tags here and let the server flush them (via +/// [`take_dropped_plugin_warnings`]) once logging is initialized, so the warning +/// actually reaches the configured log output. +static DROPPED_PLUGIN_TAGS: Mutex> = Mutex::new(Vec::new()); + +/// Records a plugin option tag that was dropped because it is not recognized by +/// this build (for example, an enterprise plugin option seen by an open-source +/// build). +pub fn record_dropped_plugin(tag: &str) { + if let Ok(mut tags) = DROPPED_PLUGIN_TAGS.lock() { + tags.push(tag.to_string()); + } +} + +/// Takes (and clears) the buffered dropped-plugin tags so they can be logged +/// after the global logging is up. +pub fn take_dropped_plugin_warnings() -> Vec { + DROPPED_PLUGIN_TAGS + .lock() + .map(|mut tags| std::mem::take(&mut *tags)) + .unwrap_or_default() +} + +/// Deserializes a list of plugin options leniently. +/// +/// Each entry is normally a single-key `{"tag": payload}` object, but a +/// *multi-key* object is also accepted and split: every top-level key is +/// treated as an independent plugin option. Keys naming a *known* variant are +/// deserialized and kept; keys naming a variant this build doesn't recognize are +/// dropped (and buffered via [`record_dropped_plugin`] for deferred logging). +/// This lets a single config file be shared across builds that compile different +/// sets of plugins — including a table that mixes a known plugin with unknown +/// ones — without aborting startup. +/// +/// A *known* variant whose payload is genuinely malformed still propagates its +/// error, so a misconfigured plugin is never silently disabled; only genuinely +/// unknown options are ignored. For privacy only the unrecognized variant *tag* +/// is recorded, never the raw payload (which may contain secrets). +/// +/// This is generic over `T` so the open-source `plugins` crate and the +/// enterprise `ent-plugins` crate share a single implementation without either +/// having to enumerate its variants (see [`is_unknown_variant_tag`]). +pub fn filter_known_plugin_options( + values: Vec, +) -> Result, serde_json::Error> { + let mut out = Vec::with_capacity(values.len()); + for value in values { + collect_from_entry::(&value, &mut out)?; + } + Ok(out) +} + +/// Extracts the recognized plugin options from one config entry. +/// +/// A normal entry is a single-key `{"tag": payload}` object. We additionally +/// accept a *multi-key* object and treat every top-level key as an independent +/// plugin option: each key naming a *known* variant is deserialized (and a +/// malformed payload for a known variant still errors, so a misconfigured plugin +/// is never silently disabled), while keys naming unknown variants are dropped +/// with a warning. A non-object entry (e.g. a bare number) is deserialized +/// directly and surfaces any error. +fn collect_from_entry( + value: &Value, + out: &mut Vec, +) -> Result<(), serde_json::Error> { + let Some(map) = value.as_object() else { + // Not a tagged object (e.g. a bare number); a bare string can still + // deserialize a unit variant. Surface any error. + out.push(T::deserialize(value)?); + return Ok(()); + }; + for (tag, payload) in map { + if is_unknown_variant_tag::(tag) { + record_dropped_plugin(tag); + continue; + } + // Known variant: deserialize just `{"tag": payload}` so a malformed + // payload is reported against this variant rather than swallowed. + let mut single = serde_json::Map::new(); + single.insert(tag.clone(), payload.clone()); + out.push(T::deserialize(&Value::Object(single))?); + } + Ok(()) +} + +/// Decides whether `tag` names a variant unknown to `T`, *without* `T` having +/// to enumerate its variants. +/// +/// Two probe payloads (`{"tag": null}` and `{"tag": true}`) are re-deserialized +/// for the same tag: +/// - If `T` accepts either payload, the tag is recognized (a known variant). +/// - If both are rejected, the tag is unknown iff the two errors are identical. +/// An unknown variant yields a payload-independent "unknown variant" error for +/// both probes, whereas a known variant rejects `null` and `bool` with +/// *different* "invalid type" messages. +fn is_unknown_variant_tag(tag: &str) -> bool { + let probe = |payload: Value| { + let mut map = serde_json::Map::new(); + map.insert(tag.to_string(), payload); + Value::Object(map) + }; + match ( + T::deserialize(&probe(Value::Null)), + T::deserialize(&probe(Value::Bool(true))), + ) { + (Ok(_), _) | (_, Ok(_)) => false, + (Err(a), Err(b)) => a.to_string() == b.to_string(), + } +} + +#[cfg(test)] +mod tests { + use std::sync::{Mutex, OnceLock}; + + use serde::Deserialize; + + use super::*; + + // The dropped-tag buffer is process-global, so every test in this module + // serializes through this lock to keep buffer assertions deterministic. + fn lock() -> std::sync::MutexGuard<'static, ()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())).lock().unwrap() + } + + #[derive(Debug, PartialEq, Deserialize)] + struct UnitPayload; + + /// A stand-in for a real `PluginOptions` enum covering the variant shapes + /// that occur in practice (newtype-over-unit, struct, newtype-over-scalar). + #[derive(Debug, PartialEq, Deserialize)] + enum DummyPlugins { + Unit(UnitPayload), + Struct { x: u32 }, + Newtype(String), + } + + #[test] + fn detects_unknown_vs_known_tags() { + let _g = lock(); + // `is_unknown_variant_tag` never touches the dropped-tag buffer. + for tag in ["Bogus", "DoesNotExist", "NotARealVariant"] { + assert!( + is_unknown_variant_tag::(tag), + "{tag} should be unknown" + ); + } + for tag in ["Unit", "Struct", "Newtype"] { + assert!( + !is_unknown_variant_tag::(tag), + "{tag} should be known" + ); + } + } + + #[test] + fn filter_drops_unknown_keeps_known() { + let _g = lock(); + let _ = take_dropped_plugin_warnings(); + let values: Vec = serde_json::from_str( + r#"[ + {"Unit": null}, + {"Bogus": {"a": 1}}, + {"AnotherUnknown": {}}, + {"Struct": {"x": 7}}, + {"Newtype": "hi"} + ]"#, + ) + .unwrap(); + let kept = filter_known_plugin_options::(values).unwrap(); + assert_eq!( + kept, + vec![ + DummyPlugins::Unit(UnitPayload), + DummyPlugins::Struct { x: 7 }, + DummyPlugins::Newtype("hi".to_string()), + ] + ); + } + + #[test] + fn filter_errors_on_malformed_known_variant() { + let _g = lock(); + // struct/newtype variants cannot come from these payloads, so a *known* + // variant's malformed payload must error (not be silently dropped). + for bad in [ + r#"[{"Struct": 5}]"#, + r#"[{"Struct": null}]"#, + r#"[{"Struct": "oops"}]"#, + r#"[{"Newtype": 7}]"#, + ] { + let values: Vec = serde_json::from_str(bad).unwrap(); + assert!( + filter_known_plugin_options::(values).is_err(), + "expected error for {bad}" + ); + } + } + + #[test] + fn filter_errors_on_non_tagged_shape() { + let _g = lock(); + let values: Vec = serde_json::from_str(r#"[123]"#).unwrap(); + assert!(filter_known_plugin_options::(values).is_err()); + } + + #[test] + fn filter_drops_multi_key_all_unknown_entry() { + let _g = lock(); + let _ = take_dropped_plugin_warnings(); + // A single entry carrying several *unknown* keys must be dropped, not + // error with serde's "expected map with a single key". + let values: Vec = + serde_json::from_str(r#"[{"unknown_one": 1, "unknown_two": 2}]"#).unwrap(); + let kept = filter_known_plugin_options::(values).unwrap(); + assert!(kept.is_empty()); + let mut recorded = take_dropped_plugin_warnings(); + recorded.sort(); + assert_eq!( + recorded, + vec!["unknown_one".to_string(), "unknown_two".to_string()] + ); + } + + #[test] + fn filter_keeps_known_drops_unknown_in_mixed_entry() { + let _g = lock(); + let _ = take_dropped_plugin_warnings(); + // A table mixing a *known* plugin with unknown ones keeps the known + // plugin and drops (warns) the unknown ones — it must not error. + let values: Vec = + serde_json::from_str(r#"[{"Struct": {"x": 1}, "bogus": 2}]"#).unwrap(); + let kept = filter_known_plugin_options::(values).unwrap(); + assert_eq!(kept, vec![DummyPlugins::Struct { x: 1 }]); + assert_eq!(take_dropped_plugin_warnings(), vec!["bogus".to_string()]); + } + + #[test] + fn filter_errors_on_malformed_known_in_mixed_entry() { + let _g = lock(); + let _ = take_dropped_plugin_warnings(); + // A *known* variant with a genuinely malformed payload still errors, + // even when an unknown key sits next to it in the same table. + let values: Vec = serde_json::from_str(r#"[{"Struct": 5, "bogus": 2}]"#).unwrap(); + assert!(filter_known_plugin_options::(values).is_err()); + // The malformed known variant aborted before the unknown was recorded. + assert!(take_dropped_plugin_warnings().is_empty()); + } + + #[test] + fn dropped_tags_are_buffered_with_tag_only() { + let _g = lock(); + let _ = take_dropped_plugin_warnings(); + // Privacy: only the tag is recorded, never the raw payload. + let values: Vec = + serde_json::from_str(r#"[{"SecretPlugin": {"token": "hunter2"}}]"#).unwrap(); + filter_known_plugin_options::(values).unwrap(); + assert_eq!( + take_dropped_plugin_warnings(), + vec!["SecretPlugin".to_string()] + ); + } +} diff --git a/src/plugins/src/options.rs b/src/plugins/src/options.rs index 5af0c4ab3a..b88e577a04 100644 --- a/src/plugins/src/options.rs +++ b/src/plugins/src/options.rs @@ -14,6 +14,7 @@ use common_options::plugin_options::{PluginOptionsDeserializer, PluginOptionsSerializer}; use serde::{Deserialize, Serialize}; +use serde_json::Value; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct DummyOptions; @@ -34,6 +35,7 @@ impl PluginOptionsSerializer for PluginOptionsList { } } } + pub struct PluginOptionsDeserializerImpl; impl PluginOptionsDeserializer> for PluginOptionsDeserializerImpl { @@ -41,7 +43,99 @@ impl PluginOptionsDeserializer> for PluginOptionsDeserializer if payload.is_empty() { Ok(vec![]) } else { - serde_json::from_str(payload) + let values: Vec = serde_json::from_str(payload)?; + common_options::plugin_options::filter_known_plugin_options(values) } } } + +#[cfg(test)] +mod tests { + use std::sync::{Mutex, OnceLock}; + + use super::*; + + // The dropped-tag buffer is process-global, so tests that touch it run serialized. + fn lock() -> std::sync::MutexGuard<'static, ()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())).lock().unwrap() + } + + fn filter(values: Vec) -> Result, serde_json::Error> { + common_options::plugin_options::filter_known_plugin_options(values) + } + + #[test] + fn test_drops_unknown_keeps_known() { + let _g = lock(); + // `null` is the serialized form of the unit-struct `DummyOptions`. + let payload = r#"[ + {"Dummy": null}, + {"SomeEnterpriseThing": {"foo": "bar", "num": 5}}, + {"AnotherUnknown": {}} + ]"#; + let values: Vec = serde_json::from_str(payload).unwrap(); + let kept = filter(values).unwrap(); + assert_eq!(kept, vec![PluginOptions::Dummy(DummyOptions)]); + // Unknown tags were buffered for deferred logging. + assert_eq!( + common_options::plugin_options::take_dropped_plugin_warnings(), + vec![ + "SomeEnterpriseThing".to_string(), + "AnotherUnknown".to_string() + ] + ); + } + + #[test] + fn test_unknown_variant_tag_is_not_in_payload_warning() { + let _g = lock(); + // Privacy (#5): only the tag is recorded, never the raw payload. + let values: Vec = + serde_json::from_str(r#"[{"SecretPlugin": {"token": "hunter2"}}]"#).unwrap(); + filter(values).unwrap(); + let recorded = common_options::plugin_options::take_dropped_plugin_warnings(); + assert_eq!(recorded, vec!["SecretPlugin".to_string()]); + } + + #[test] + fn test_known_variant_with_malformed_payload_errors() { + let _g = lock(); + // Regression (#3): a known variant with a bad payload must NOT be + // silently dropped. + for bad in [r#"[{"Dummy": 1}]"#, r#"[{"Dummy": {"x": 1}}]"#] { + let values: Vec = serde_json::from_str(bad).unwrap(); + let err = filter(values).unwrap_err(); + assert!( + err.to_string().contains("Dummy"), + "expected the error to reference the Dummy variant, got: {err}" + ); + } + // Nothing should have been buffered. + assert!(common_options::plugin_options::take_dropped_plugin_warnings().is_empty()); + } + + #[test] + fn test_unrecognized_shape_errors() { + let _g = lock(); + // Not a recognizable plugin entry: surface the error. + let values: Vec = serde_json::from_str(r#"[123]"#).unwrap(); + assert!(filter(values).is_err()); + } + + #[test] + fn test_deserializer_impl_is_lenient_for_unknown() { + let _g = lock(); + let payload = r#"[{"Dummy": null}, {"UnknownPlugin": {"x": 1}}]"#; + let kept = PluginOptionsDeserializerImpl.deserialize(payload).unwrap(); + assert_eq!(kept, vec![PluginOptions::Dummy(DummyOptions)]); + common_options::plugin_options::take_dropped_plugin_warnings(); + } + + #[test] + fn test_deserializer_impl_errors_on_bad_known_payload() { + let _g = lock(); + let payload = r#"[{"Dummy": 1}]"#; + assert!(PluginOptionsDeserializerImpl.deserialize(payload).is_err()); + } +}