Merge branch 'vectordotdev:master' into master

Author: handlerbot

.github/workflows/publish.yml

@@ -133,6 +133,7 @@ members = [
 ]
 
 [workspace.dependencies]
+cfg-if = { version = "1.0.0", default-features = false }
 chrono = { version = "0.4.38", default-features = false, features = ["clock", "serde"] }
 chrono-tz = { version = "0.10.0", default-features = false, features = ["serde"] }
 clap = { version = "4.5.20", default-features = false, features = ["derive", "error-context", "env", "help", "std", "string", "usage", "wrap_help"] }

Cargo.lock

@@ -70,6 +70,7 @@ vector-config = { path = "../vector-config" }
 vector-config-common = { path = "../vector-config-common" }
 vector-config-macros = { path = "../vector-config-macros" }
 vrl.workspace = true
+cfg-if.workspace = true
 
 [target.'cfg(target_os = "macos")'.dependencies]
 security-framework = "2.10.0"

Cargo.toml

@@ -30,6 +30,7 @@ pub const MEMORY_BUFFER_DEFAULT_MAX_EVENTS: NonZeroUsize =
 // This enum should be kept alphabetically sorted as the bitmask value is used when
 // sorting sources by data type in the GraphQL API.
 #[bitmask(u8)]
+#[bitmask_config(flags_iter)]
 pub enum DataType {
     Log,
     Metric,
@@ -38,11 +39,11 @@ pub enum DataType {
 
 impl fmt::Display for DataType {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        let mut t = Vec::new();
-        self.contains(DataType::Log).then(|| t.push("Log"));
-        self.contains(DataType::Metric).then(|| t.push("Metric"));
-        self.contains(DataType::Trace).then(|| t.push("Trace"));
-        f.write_str(&t.join(","))
+        f.debug_list()
+            .entries(
+                Self::flags().filter_map(|&(name, value)| self.contains(value).then_some(name)),
+            )
+            .finish()
     }
 }
 
@@ -191,6 +192,24 @@ impl SourceOutput {
     }
 }
 
+fn fmt_helper(
+    f: &mut fmt::Formatter<'_>,
+    maybe_port: Option<&String>,
+    data_type: DataType,
+) -> fmt::Result {
+    match maybe_port {
+        Some(port) => write!(f, "port: \"{port}\",",),
+        None => write!(f, "port: None,"),
+    }?;
+    write!(f, " types: {data_type}")
+}
+
+impl fmt::Display for SourceOutput {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt_helper(f, self.port.as_ref(), self.ty)
+    }
+}
+
 #[derive(Debug, Clone, PartialEq)]
 pub struct TransformOutput {
     pub port: Option<String>,
@@ -203,6 +222,12 @@ pub struct TransformOutput {
     pub log_schema_definitions: HashMap<OutputId, schema::Definition>,
 }
 
+impl fmt::Display for TransformOutput {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt_helper(f, self.port.as_ref(), self.ty)
+    }
+}
+
 impl TransformOutput {
     /// Create a `TransformOutput` of the given data type that contains multiple [`schema::Definition`]s.
     /// Designed for use in transforms.

lib/vector-core/Cargo.toml

@@ -1,32 +1,77 @@
-use std::sync::{LazyLock, OnceLock};
-
+use cfg_if::cfg_if;
 use vector_common::request_metadata::GroupedCountByteSize;
 use vector_config::configurable_component;
 
-static TELEMETRY: OnceLock<Telemetry> = OnceLock::new();
-static TELEMETRY_DEFAULT: LazyLock<Telemetry> = LazyLock::new(Telemetry::default);
-
-/// Loads the telemetry options from configurations and sets the global options.
-/// Once this is done, configurations can be correctly loaded using configured
-/// log schema defaults.
-///
-/// # Errors
-///
-/// This function will fail if the `builder` fails.
-///
-/// # Panics
-///
-/// If deny is set, will panic if telemetry has already been set.
-pub fn init_telemetry(telemetry: Telemetry, deny_if_set: bool) {
-    assert!(
-        !(TELEMETRY.set(telemetry).is_err() && deny_if_set),
-        "Couldn't set telemetry"
-    );
-}
+cfg_if! {
+    // The telemetry code assumes a process wide singleton. When running `cargo test`,
+    // multiple threads might try to read/write this global.
+    if #[cfg(any(test, feature = "test"))] {
+        use std::sync::{Arc, Mutex};
+
+        thread_local! {
+            static TELEMETRY: Arc<Mutex<Option<Telemetry>>> = Arc::new(Mutex::new(None));
+        }
+
+        /// Test implementation.
+        ///
+        /// # Panics
+        ///
+        /// If deny is set, will panic if telemetry has already been set.
+        /// Also, panics if the lock is poisoned.
+        pub fn init_telemetry(telemetry: Telemetry, deny_if_set: bool) {
+            TELEMETRY.with(|tl| {
+                let mut tl = tl.lock().expect("telemetry lock poisoned");
+                assert!(!(tl.is_some() && deny_if_set), "Couldn't set telemetry");
+                *tl = Some(telemetry);
+            });
+        }
+
+        /// Test implementation.
+        ///
+        /// # Panics
+        ///
+        /// If the lock is poisoned.
+         pub fn telemetry() -> Telemetry {
+            TELEMETRY.with(|tl| {
+               let mut tl = tl.lock().expect("telemetry lock poisoned");
+                // For non-test code we return `TELEMETRY_DEFAULT`.
+                // For test code, we will instantiate a default instance per thread.
+                if tl.is_none() {
+                    *tl = Some(Telemetry::default());
+                }
+                tl.clone().unwrap()
+            })
+        }
+    }
+    else {
+        use std::sync::{LazyLock, OnceLock};
 
-/// Returns the telemetry configuration options.
-pub fn telemetry() -> &'static Telemetry {
-    TELEMETRY.get().unwrap_or(&TELEMETRY_DEFAULT)
+        static TELEMETRY: OnceLock<Telemetry> = OnceLock::new();
+        static TELEMETRY_DEFAULT: LazyLock<Telemetry> = LazyLock::new(Telemetry::default);
+
+        /// Loads the telemetry options from configurations and sets the global options.
+        /// Once this is done, configurations can be correctly loaded using configured
+        /// log schema defaults.
+        ///
+        /// # Errors
+        ///
+        /// This function will fail if the `builder` fails.
+        ///
+        /// # Panics
+        ///
+        /// If deny is set, will panic if telemetry has already been set.
+        pub fn init_telemetry(telemetry: Telemetry, deny_if_set: bool) {
+            assert!(
+                !(TELEMETRY.set(telemetry).is_err() && deny_if_set),
+                "Couldn't set telemetry"
+            );
+        }
+
+        /// Returns the telemetry configuration options.
+        pub fn telemetry() -> &'static Telemetry {
+            TELEMETRY.get().unwrap_or(&TELEMETRY_DEFAULT)
+        }
+    }
 }
 
 /// Sets options for the telemetry that Vector emits.

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

@@ -1,10 +1,10 @@
-use indexmap::{set::IndexSet, IndexMap};
-use std::collections::{HashMap, HashSet, VecDeque};
-
 use super::{
     schema, ComponentKey, DataType, OutputId, SinkOuter, SourceOuter, SourceOutput, TransformOuter,
     TransformOutput,
 };
+use indexmap::{set::IndexSet, IndexMap};
+use std::collections::{HashMap, HashSet, VecDeque};
+use std::fmt;
 
 #[derive(Debug, Clone)]
 pub enum Node {
@@ -20,6 +20,33 @@ pub enum Node {
     },
 }
 
+impl fmt::Display for Node {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Node::Source { outputs } => {
+                write!(f, "component_kind: source\n  outputs:")?;
+                for output in outputs {
+                    write!(f, "\n    {}", output)?;
+                }
+                Ok(())
+            }
+            Node::Transform { in_ty, outputs } => {
+                write!(
+                    f,
+                    "component_kind: source\n  input_types: {in_ty}\n  outputs:"
+                )?;
+                for output in outputs {
+                    write!(f, "\n    {}", output)?;
+                }
+                Ok(())
+            }
+            Node::Sink { ty } => {
+                write!(f, "component_kind: sink\n  types: {ty}")
+            }
+        }
+    }
+}
+
 #[derive(Debug, Clone)]
 struct Edge {
     from: OutputId,
@@ -139,9 +166,16 @@ impl Graph {
                 Some(Node::Sink { .. }) => "sink",
                 _ => panic!("only transforms and sinks have inputs"),
             };
+            info!(
+                "Available components:\n{}",
+                self.nodes
+                    .iter()
+                    .map(|(key, node)| format!("\"{}\":\n  {}", key, node))
+                    .collect::<Vec<_>>()
+                    .join("\n")
+            );
             Err(format!(
-                "Input \"{}\" for {} \"{}\" doesn't match any components.",
-                from, output_type, to
+                "Input \"{from}\" for {output_type} \"{to}\" doesn't match any components.",
             ))
         }
     }
@@ -510,7 +544,7 @@ mod test {
 
         assert_eq!(
             Err(vec![
-                "Data type mismatch between in (Log) and out (Metric)".into()
+                "Data type mismatch between in ([\"Log\"]) and out ([\"Metric\"])".into()
             ]),
             graph.typecheck()
         );