chore(config): initial integration of configuration schema for sources (

#13005)

Cargo.toml

@@ -122,6 +122,8 @@ tracing-limit = { path = "lib/tracing-limit" }
 value = { path = "lib/value" }
 vector_buffers = { path = "lib/vector-buffers", default-features = false }
 vector_common = { path = "lib/vector-common" }
+vector_config = { path = "lib/vector-config" }
+vector_config_macros = { path = "lib/vector-config-macros" }
 vector_core = { path = "lib/vector-core", default-features = false, features = ["vrl"] }
 vector-api-client = { path = "lib/vector-api-client", optional = true }
 vector-vrl-functions = { path = "lib/vector-vrl-functions" }

lib/codecs/Cargo.toml

@@ -20,6 +20,8 @@ tokio-util = { version = "0.7", default-features = false, features = ["codec"] }
 tracing = { version = "0.1", default-features = false }
 value = { path = "../value", default-features = false }
 vector_common = { path = "../vector-common", default-features = false }
+vector_config = { path = "../vector-config", default-features = false }
+vector_config_macros = { path = "../vector-config-macros", default-features = false }
 vector_core = { path = "../vector-core", default-features = false }
 
 [dev-dependencies]

lib/codecs/src/decoding/framing/character_delimited.rs

@@ -3,6 +3,7 @@ use memchr::memchr;
 use serde::{Deserialize, Serialize};
 use tokio_util::codec::Decoder;
 use tracing::{trace, warn};
+use vector_config::configurable_component;
 
 use super::BoxedFramingError;
 
@@ -28,7 +29,8 @@ impl CharacterDelimitedDecoderConfig {
 }
 
 /// Options for building a `CharacterDelimitedDecoder`.
-#[derive(Debug, Clone, Deserialize, Serialize, PartialEq)]
+#[configurable_component]
+#[derive(Clone, Debug, PartialEq)]
 pub struct CharacterDelimitedDecoderOptions {
     /// The character that delimits byte sequences.
     #[serde(with = "vector_core::serde::ascii_char")]

lib/codecs/src/decoding/framing/newline_delimited.rs

@@ -2,6 +2,7 @@ use bytes::{Bytes, BytesMut};
 use derivative::Derivative;
 use serde::{Deserialize, Serialize};
 use tokio_util::codec::Decoder;
+use vector_config::configurable_component;
 
 use super::{BoxedFramingError, CharacterDelimitedDecoder};
 
@@ -17,7 +18,8 @@ pub struct NewlineDelimitedDecoderConfig {
 }
 
 /// Options for building a `NewlineDelimitedDecoder`.
-#[derive(Debug, Clone, Derivative, Deserialize, Serialize, PartialEq)]
+#[configurable_component]
+#[derive(Clone, Debug, Derivative, PartialEq)]
 #[derivative(Default)]
 pub struct NewlineDelimitedDecoderOptions {
     /// The maximum length of the byte buffer.

lib/codecs/src/decoding/framing/octet_counting.rs

@@ -5,6 +5,7 @@ use derivative::Derivative;
 use serde::{Deserialize, Serialize};
 use tokio_util::codec::{LinesCodec, LinesCodecError};
 use tracing::trace;
+use vector_config::configurable_component;
 
 use super::BoxedFramingError;
 
@@ -31,9 +32,11 @@ impl OctetCountingDecoderConfig {
 }
 
 /// Options for building a `OctetCountingDecoder`.
-#[derive(Debug, Clone, Derivative, Deserialize, Serialize, PartialEq)]
+#[configurable_component]
+#[derive(Clone, Debug, Derivative, PartialEq)]
 #[derivative(Default)]
 pub struct OctetCountingDecoderOptions {
+    /// The maximum length of the byte buffer.
     #[serde(skip_serializing_if = "vector_core::serde::skip_serializing_if_default")]
     max_length: Option<usize>,
 }

lib/codecs/src/decoding/mod.rs

@@ -23,8 +23,8 @@ pub use framing::{
     NewlineDelimitedDecoderConfig, NewlineDelimitedDecoderOptions, OctetCountingDecoder,
     OctetCountingDecoderConfig, OctetCountingDecoderOptions,
 };
-use serde::{Deserialize, Serialize};
 use smallvec::SmallVec;
+use vector_config::configurable_component;
 use vector_core::{config::DataType, event::Event, schema};
 
 /// An error that occurred while decoding structured events from a byte stream /
@@ -68,7 +68,8 @@ impl StreamDecodingError for Error {
 // Unfortunately, copying options of the nested enum variants is necessary
 // since `serde` doesn't allow `flatten`ing these:
 // https://github.com/serde-rs/serde/issues/1402.
-#[derive(Debug, Clone, Deserialize, Serialize)]
+#[configurable_component]
+#[derive(Clone, Debug)]
 #[serde(tag = "method", rename_all = "snake_case")]
 pub enum FramingConfig {
     /// Configures the `BytesDecoder`.
@@ -216,7 +217,8 @@ impl tokio_util::codec::Decoder for Framer {
 // Unfortunately, copying options of the nested enum variants is necessary
 // since `serde` doesn't allow `flatten`ing these:
 // https://github.com/serde-rs/serde/issues/1402.
-#[derive(Debug, Clone, Deserialize, Serialize)]
+#[configurable_component]
+#[derive(Clone, Debug)]
 #[serde(tag = "codec", rename_all = "snake_case")]
 pub enum DeserializerConfig {
     /// Configures the `BytesDeserializer`.

lib/vector-common/Cargo.toml

@@ -56,3 +56,4 @@ smallvec = { version = "1", default-features = false }
 snafu = { version = "0.7", optional = true }
 tracing = { version = "0.1.34", default-features = false }
 value = { path = "../value", features = ["json"] }
+vector_config = { path = "../vector-config" }

lib/vector-common/src/datetime.rs

@@ -3,6 +3,11 @@ use std::fmt::Debug;
 use chrono::{DateTime, Local, ParseError, TimeZone as _, Utc};
 use chrono_tz::Tz;
 use derivative::Derivative;
+use vector_config::{
+    schema::{finalize_schema, generate_string_schema},
+    schemars::{gen::SchemaGenerator, schema::SchemaObject},
+    Configurable, Metadata,
+};
 
 #[derive(Clone, Copy, Debug, Derivative, Eq, PartialEq)]
 #[derivative(Default)]
@@ -82,3 +87,19 @@ pub mod ser_de {
         }
     }
 }
+
+impl<'de> Configurable<'de> for TimeZone {
+    fn referencable_name() -> Option<&'static str> {
+        Some("vector_common::TimeZone")
+    }
+
+    fn description() -> Option<&'static str> {
+        Some("Strongly-typed list of timezones as defined in the `tz` database.")
+    }
+
+    fn generate_schema(gen: &mut SchemaGenerator, overrides: Metadata<'de, Self>) -> SchemaObject {
+        let mut schema = generate_string_schema();
+        finalize_schema(gen, &mut schema, overrides);
+        schema
+    }
+}

lib/vector-config-common/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "vector-config-common"
+name = "vector_config_common"
 version = "0.1.0"
 edition = "2021"
 

lib/vector-config-macros/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "vector-config-macros"
+name = "vector_config_macros"
 version = "0.1.0"
 edition = "2021"
 
@@ -12,4 +12,4 @@ proc-macro2 = { version = "1.0", default-features = false }
 quote = { version = "1.0", default-features = false }
 serde_derive_internals = "0.26"
 syn = { version = "1.0", default-features = false }
-vector-config-common = { path = "../vector-config-common" }
+vector_config_common = { path = "../vector-config-common" }

lib/vector-config-macros/src/ast/container.rs

@@ -1,4 +1,8 @@
-use darling::{error::Accumulator, util::path_to_string, FromAttributes, FromMeta};
+use darling::{
+    error::Accumulator,
+    util::{path_to_string, Flag},
+    FromAttributes, FromMeta,
+};
 use serde_derive_internals::{ast as serde_ast, Ctxt, Derive};
 use syn::{DeriveInput, ExprPath, Generics, Ident, NestedMeta};
 
@@ -11,10 +15,10 @@ use super::{
 };
 
 const ERR_NO_ENUM_TUPLES: &str = "enum variants cannot be tuples (multiple unnamed fields)";
-const ERR_NO_ENUM_NEWTYPE_INTERNAL_TAG: &str = "newtype variants (i.e. `enum SomeEnum { SomeVariant(T) }`) cannot be used with tag-only mode as the type inside may or may not support embedding the tag field";
 const ERR_NO_ENUM_VARIANT_DESCRIPTION: &str = "enum variants must have a description i.e. `/// This is a description` or `#[configurable(description = \"This is a description...\")]`";
 const ERR_ENUM_UNTAGGED_DUPLICATES: &str = "enum variants must be unique in style/shape when in untagged mode i.e. there cannot be multiple unit variants, or tuple variants with the same fields, etc";
 const ERR_NO_UNIT_STRUCTS: &str = "unit structs are not supported by `Configurable`";
+const ERR_MISSING_DESC: &str = "all structs/enums must have a description i.e. `/// This is a description` or `#[configurable(description = \"This is a description...\")]`";
 
 pub struct Container<'a> {
     original: &'a DeriveInput,
@@ -70,16 +74,6 @@ impl<'a> Container<'a> {
                                 );
                             }
 
-                            // We don't support internal tag for newtype variants, because `serde` doesn't support it.
-                            if variant.style() == Style::Newtype
-                                && matches!(variant.tagging(), Tagging::Internal { .. })
-                            {
-                                accumulator.push(
-                                    darling::Error::custom(ERR_NO_ENUM_NEWTYPE_INTERNAL_TAG)
-                                        .with_span(variant),
-                                );
-                            }
-
                             // All variants must have a description.  No derived/transparent mode.
                             if variant.description().is_none() {
                                 accumulator.push(
@@ -127,6 +121,15 @@ impl<'a> Container<'a> {
                     },
                 };
 
+                // All containers must have a description: no ifs, ands, or buts.
+                //
+                // The compile-time errors are a bit too inscrutable otherwise, and inscrutable errors are not very
+                // helpful when using procedural macros.
+                if attrs.description.is_none() {
+                    accumulator
+                        .push(darling::Error::custom(ERR_MISSING_DESC).with_span(&serde.ident));
+                }
+
                 let original = input;
                 let name = serde.attrs.name().deserialize_name();
                 let default_value = get_serde_default_value(serde.attrs.default());
@@ -172,7 +175,7 @@ impl<'a> Container<'a> {
     }
 
     pub fn deprecated(&self) -> bool {
-        self.attrs.deprecated
+        self.attrs.deprecated.is_present()
     }
 
     pub fn metadata(&self) -> impl Iterator<Item = &(String, String)> {
@@ -189,19 +192,13 @@ impl<'a> Container<'a> {
 struct Attributes {
     title: Option<String>,
     description: Option<String>,
-    #[darling(skip)]
-    deprecated: bool,
+    deprecated: Flag,
     #[darling(multiple)]
     metadata: Vec<Metadata>,
 }
 
 impl Attributes {
     fn finalize(mut self, forwarded_attrs: &[syn::Attribute]) -> darling::Result<Self> {
-        // Parse any forwarded attributes that `darling` left us.
-        self.deprecated = forwarded_attrs
-            .iter()
-            .any(|a| a.path.is_ident("deprecated"));
-
         // We additionally attempt to extract a title/description from the forwarded doc attributes, if they exist.
         // Whether we extract both a title and description, or just description, is documented in more detail in
         // `try_extract_doc_title_description` itself.