feat(config): add wildcard_matching config option (#23011)

* add relaxed_wildcart_matching config option

* allow empty results for glob inputs if relaxed_wildcard_matching is enabled

* optimize condition

* fix fmt

* fix clippy

* remove option

* fix fmt

* Revert "remove option"

This reverts commit 901cf08.

* generate docs

* add changelog fragment

* fix typo

* add tests

* add config example to changelog

* use enum instead of boolean

* regenerate docs

* update changelog

Author: simplepad

changelog.d/23011_add_relaxed_wildcard_matching.feature.md

@@ -0,0 +1,24 @@
+Added `wildcard_matching` global config option to set wildcard matching mode for inputs. Relaxed mode allows configurations with wildcards that do not match any inputs to be accepted without causing an error.
+
+Example config:
+
+```yaml
+wildcard_matching: relaxed
+
+sources:
+  stdin:
+    type: stdin
+
+# note - no transforms
+
+sinks:
+  stdout:
+    type: console
+    encoding:
+      codec: json
+    inputs:
+      - "runtime-added-transform-*"
+
+```
+
+authors: simplepad

lib/vector-core/src/config/global_options.rs

@@ -31,6 +31,19 @@ pub(crate) enum DataDirError {
     },
 }
 
+/// Specifies the wildcard matching mode, relaxed allows configurations where wildcard doesn not match any existing inputs
+#[configurable_component]
+#[derive(Clone, Debug, Copy, PartialEq, Eq, Default)]
+#[serde(rename_all = "lowercase")]
+pub enum WildcardMatching {
+    /// Strict matching (must match at least one existing input)
+    #[default]
+    Strict,
+
+    /// Relaxed matching (must match 0 or more inputs)
+    Relaxed,
+}
+
 /// Global configuration options.
 //
 // If this is modified, make sure those changes are reflected in the `ConfigBuilder::append`
@@ -48,6 +61,14 @@ pub struct GlobalOptions {
     #[configurable(metadata(docs::common = false))]
     pub data_dir: Option<PathBuf>,
 
+    /// Set wildcard matching mode for inputs
+    ///
+    /// Setting this to "relaxed" allows configurations with wildcards that do not match any inputs
+    /// to be accepted without causing an error.
+    #[serde(skip_serializing_if = "crate::serde::is_default")]
+    #[configurable(metadata(docs::common = false, docs::required = false))]
+    pub wildcard_matching: Option<WildcardMatching>,
+
     /// Default log schema for all events.
     ///
     /// This is used if a component does not have its own specific log schema. All events use a log
@@ -182,6 +203,13 @@ impl GlobalOptions {
     pub fn merge(&self, with: Self) -> Result<Self, Vec<String>> {
         let mut errors = Vec::new();
 
+        if conflicts(
+            self.wildcard_matching.as_ref(),
+            with.wildcard_matching.as_ref(),
+        ) {
+            errors.push("conflicting values for 'wildcard_matching' found".to_owned());
+        }
+
         if conflicts(self.proxy.http.as_ref(), with.proxy.http.as_ref()) {
             errors.push("conflicting values for 'proxy.http' found".to_owned());
         }
@@ -250,6 +278,7 @@ impl GlobalOptions {
         if errors.is_empty() {
             Ok(Self {
                 data_dir,
+                wildcard_matching: self.wildcard_matching.or(with.wildcard_matching),
                 log_schema,
                 telemetry,
                 acknowledgements: self.acknowledgements.merge_default(&with.acknowledgements),

lib/vector-core/src/config/mod.rs

@@ -13,7 +13,7 @@ pub mod proxy;
 mod telemetry;
 
 use crate::event::LogEvent;
-pub use global_options::GlobalOptions;
+pub use global_options::{GlobalOptions, WildcardMatching};
 pub use log_schema::{init_log_schema, log_schema, LogSchema};
 use lookup::{lookup_v2::ValuePath, path, PathPrefix};
 pub use output_id::OutputId;

lib/vector-lib/src/lib.rs

@@ -34,7 +34,7 @@ pub mod config {
         clone_input_definitions, init_log_schema, init_telemetry, log_schema, proxy, telemetry,
         AcknowledgementsConfig, DataType, GlobalOptions, Input, LegacyKey, LogNamespace, LogSchema,
         OutputId, SourceAcknowledgementsConfig, SourceOutput, Tags, Telemetry, TransformOutput,
-        MEMORY_BUFFER_DEFAULT_MAX_EVENTS,
+        WildcardMatching, MEMORY_BUFFER_DEFAULT_MAX_EVENTS,
     };
 }
 

src/config/compiler.rs

@@ -71,7 +71,13 @@ pub fn compile(mut builder: ConfigBuilder) -> Result<(Config, Vec<String>), Vec<
         )
         .collect::<IndexMap<_, _>>();
 
-    let graph = match Graph::new(&sources_and_table_sources, &transforms, &all_sinks, schema) {
+    let graph = match Graph::new(
+        &sources_and_table_sources,
+        &transforms,
+        &all_sinks,
+        schema,
+        global.wildcard_matching.unwrap_or_default(),
+    ) {
         Ok(graph) => graph,
         Err(graph_errors) => {
             errors.extend(graph_errors);

src/config/graph.rs

@@ -1,6 +1,6 @@
 use super::{
     schema, ComponentKey, DataType, OutputId, SinkOuter, SourceOuter, SourceOutput, TransformOuter,
-    TransformOutput,
+    TransformOutput, WildcardMatching,
 };
 use indexmap::{set::IndexSet, IndexMap};
 use std::collections::{HashMap, HashSet, VecDeque};
@@ -65,17 +65,20 @@ impl Graph {
         transforms: &IndexMap<ComponentKey, TransformOuter<String>>,
         sinks: &IndexMap<ComponentKey, SinkOuter<String>>,
         schema: schema::Options,
+        wildcard_matching: WildcardMatching,
     ) -> Result<Self, Vec<String>> {
-        Self::new_inner(sources, transforms, sinks, false, schema)
+        Self::new_inner(sources, transforms, sinks, false, schema, wildcard_matching)
     }
 
     pub fn new_unchecked(
         sources: &IndexMap<ComponentKey, SourceOuter>,
         transforms: &IndexMap<ComponentKey, TransformOuter<String>>,
         sinks: &IndexMap<ComponentKey, SinkOuter<String>>,
         schema: schema::Options,
+        wildcard_matching: WildcardMatching,
     ) -> Self {
-        Self::new_inner(sources, transforms, sinks, true, schema).expect("errors ignored")
+        Self::new_inner(sources, transforms, sinks, true, schema, wildcard_matching)
+            .expect("errors ignored")
     }
 
     fn new_inner(
@@ -84,6 +87,7 @@ impl Graph {
         sinks: &IndexMap<ComponentKey, SinkOuter<String>>,
         ignore_errors: bool,
         schema: schema::Options,
+        wildcard_matching: WildcardMatching,
     ) -> Result<Self, Vec<String>> {
         let mut graph = Graph::default();
         let mut errors = Vec::new();
@@ -127,15 +131,15 @@ impl Graph {
 
         for (id, config) in transforms.iter() {
             for input in config.inputs.iter() {
-                if let Err(e) = graph.add_input(input, id, &available_inputs) {
+                if let Err(e) = graph.add_input(input, id, &available_inputs, wildcard_matching) {
                     errors.push(e);
                 }
             }
         }
 
         for (id, config) in sinks {
             for input in config.inputs.iter() {
-                if let Err(e) = graph.add_input(input, id, &available_inputs) {
+                if let Err(e) = graph.add_input(input, id, &available_inputs, wildcard_matching) {
                     errors.push(e);
                 }
             }
@@ -153,6 +157,7 @@ impl Graph {
         from: &str,
         to: &ComponentKey,
         available_inputs: &HashMap<String, OutputId>,
+        wildcard_matching: WildcardMatching,
     ) -> Result<(), String> {
         if let Some(output_id) = available_inputs.get(from) {
             self.edges.push(Edge {
@@ -166,6 +171,18 @@ impl Graph {
                 Some(Node::Sink { .. }) => "sink",
                 _ => panic!("only transforms and sinks have inputs"),
             };
+            // allow empty result if relaxed wildcard matching is enabled
+            match wildcard_matching {
+                WildcardMatching::Relaxed => {
+                    // using value != glob::Pattern::escape(value) to check if value is a glob
+                    // TODO: replace with proper check when https://github.com/rust-lang/glob/issues/72 is resolved
+                    if from != glob::Pattern::escape(from) {
+                        info!("Input \"{from}\" for {output_type} \"{to}\" didn’t match any components, but this was ignored because `relaxed_wildcard_matching` is enabled.");
+                        return Ok(());
+                    }
+                }
+                WildcardMatching::Strict => {}
+            }
             info!(
                 "Available components:\n{}",
                 self.nodes
@@ -472,9 +489,14 @@ mod test {
             }
         }
 
-        fn test_add_input(&mut self, node: &str, input: &str) -> Result<(), String> {
+        fn test_add_input(
+            &mut self,
+            node: &str,
+            input: &str,
+            wildcard_matching: WildcardMatching,
+        ) -> Result<(), String> {
             let available_inputs = self.input_map().unwrap();
-            self.add_input(input, &node.into(), &available_inputs)
+            self.add_input(input, &node.into(), &available_inputs, wildcard_matching)
         }
     }
 
@@ -655,14 +677,22 @@ mod test {
         // make sure we're good with dotted paths
         assert_eq!(
             Ok(()),
-            graph.test_add_input("errored_log_sink", "log_to_log.errors")
+            graph.test_add_input(
+                "errored_log_sink",
+                "log_to_log.errors",
+                WildcardMatching::Strict
+            )
         );
 
         // make sure that we're not cool with an unknown dotted path
         let expected = "Input \"log_to_log.not_errors\" for sink \"bad_log_sink\" doesn't match any components.".to_string();
         assert_eq!(
             Err(expected),
-            graph.test_add_input("bad_log_sink", "log_to_log.not_errors")
+            graph.test_add_input(
+                "bad_log_sink",
+                "log_to_log.not_errors",
+                WildcardMatching::Strict
+            )
         );
     }
 
@@ -745,6 +775,40 @@ mod test {
         );
     }
 
+    #[test]
+    fn wildcard_matching() {
+        let mut graph = Graph::default();
+        graph.add_source("log_source", DataType::Log);
+
+        // don't add inputs to these yet since they're not validated via these helpers
+        graph.add_sink("sink", DataType::Log, vec![]);
+
+        // make sure we're not good with non existing inputs with relaxed wildcard matching disabled
+        let wildcard_matching = WildcardMatching::Strict;
+        let expected =
+            "Input \"bad_source-*\" for sink \"sink\" doesn't match any components.".to_string();
+        assert_eq!(
+            Err(expected),
+            graph.test_add_input("sink", "bad_source-*", wildcard_matching)
+        );
+
+        // make sure we're good with non existing inputs with relaxed wildcard matching enabled
+        let wildcard_matching = WildcardMatching::Relaxed;
+        assert_eq!(
+            Ok(()),
+            graph.test_add_input("sink", "bad_source-*", wildcard_matching)
+        );
+
+        // make sure we're not good with non existing inputs that are not wildcards even when relaxed wildcard matching is enabled
+        let wildcard_matching = WildcardMatching::Relaxed;
+        let expected =
+            "Input \"bad_source-1\" for sink \"sink\" doesn't match any components.".to_string();
+        assert_eq!(
+            Err(expected),
+            graph.test_add_input("sink", "bad_source-1", wildcard_matching)
+        );
+    }
+
     #[test]
     fn paths_to_sink_simple() {
         let mut graph = Graph::default();

src/config/mod.rs

@@ -21,7 +21,7 @@ use serde::Serialize;
 use vector_config::configurable_component;
 pub use vector_lib::config::{
     AcknowledgementsConfig, DataType, GlobalOptions, Input, LogNamespace,
-    SourceAcknowledgementsConfig, SourceOutput, TransformOutput,
+    SourceAcknowledgementsConfig, SourceOutput, TransformOutput, WildcardMatching,
 };
 pub use vector_lib::configurable::component::{
     GenerateConfig, SinkDescription, TransformDescription,

src/config/unit_test/mod.rs

@@ -385,6 +385,10 @@ async fn build_unit_test(
         &transform_only_config.transforms,
         &transform_only_config.sinks,
         transform_only_config.schema,
+        transform_only_config
+            .global
+            .wildcard_matching
+            .unwrap_or_default(),
     );
     let test = test.resolve_outputs(&transform_only_graph)?;
 
@@ -401,6 +405,7 @@ async fn build_unit_test(
         &config_builder.transforms,
         &config_builder.sinks,
         config_builder.schema,
+        config_builder.global.wildcard_matching.unwrap_or_default(),
     );
 
     let mut valid_components = get_relevant_test_components(
@@ -432,6 +437,7 @@ async fn build_unit_test(
         &config_builder.transforms,
         &config_builder.sinks,
         config_builder.schema,
+        config_builder.global.wildcard_matching.unwrap_or_default(),
     );
     let valid_inputs = graph.input_map()?;
     for (_, transform) in config_builder.transforms.iter_mut() {

website/cue/reference/base/configuration.cue

@@ -901,4 +901,18 @@ base: configuration: configuration: {
 		required: false
 		type: string: examples: ["local", "America/New_York", "EST5EDT"]
 	}
+	wildcard_matching: {
+		common: false
+		description: """
+			Set wildcard matching mode for inputs
+
+			Setting this to "relaxed" allows configurations with wildcards that do not match any inputs
+			to be accepted without causing an error.
+			"""
+		required: false
+		type: string: enum: {
+			relaxed: "Relaxed matching (must match 0 or more inputs)"
+			strict:  "Strict matching (must match at least one existing input)"
+		}
+	}
 }