fix(config): Avoid parsing configuration files without interpolating …

…secrets (vectordotdev#20985) * fix(config): Avoid parsing configuration files without secrets Which caused vectordotdev#20974 Reverts: vectordotdev#17759 Signed-off-by: Jesse Szwedko <[email protected]> * Update configuration structs to default to log_schema fields after deserialization Signed-off-by: Jesse Szwedko <[email protected]> * Add changelog entry Signed-off-by: Jesse Szwedko <[email protected]> * Regenerate docs Signed-off-by: Jesse Szwedko <[email protected]> --------- Signed-off-by: Jesse Szwedko <[email protected]>
frankh · Aug 12, 2024 · 9e9689a · 9e9689a
1 parent 206e054
commit 9e9689a
Show file tree

Hide file tree

Showing 17 changed files with 91 additions and 90 deletions.
diff --git a/changelog.d/defer-log-schema-defaults.fix.md b/changelog.d/defer-log-schema-defaults.fix.md
@@ -0,0 +1 @@
+Vector no longer panics during configuration loading if a secret is used for a configuration option that has additional validation (e.g. URIs).
diff --git a/src/app.rs b/src/app.rs
@@ -489,9 +489,6 @@ pub async fn load_configs(
         paths = ?config_paths.iter().map(<&PathBuf>::from).collect::<Vec<_>>()
     );
 
-    // config::init_log_schema should be called before initializing sources.
-    config::init_log_schema(&config_paths, true).map_err(handle_config_errors)?;
-
     let mut config = config::load_from_paths_with_provider_and_secrets(
         &config_paths,
         signal_handler,
@@ -500,6 +497,7 @@ pub async fn load_configs(
     .await
     .map_err(handle_config_errors)?;
 
+    config::init_log_schema(config.global.log_schema.clone(), true);
     config::init_telemetry(config.global.telemetry.clone(), true);
 
     if !config.healthchecks.enabled {

diff --git a/src/config/mod.rs b/src/config/mod.rs
@@ -64,19 +64,9 @@ pub use unit_test::{build_unit_tests, build_unit_tests_main, UnitTestResult};
 pub use validation::warnings;
 pub use vars::{interpolate, ENVIRONMENT_VARIABLE_INTERPOLATION_REGEX};
 pub use vector_lib::config::{
-    init_telemetry, log_schema, proxy::ProxyConfig, telemetry, LogSchema, OutputId,
+    init_log_schema, init_telemetry, log_schema, proxy::ProxyConfig, telemetry, LogSchema, OutputId,
 };
 
-/// Loads Log Schema from configurations and sets global schema.
-/// Once this is done, configurations can be correctly loaded using
-/// configured log schema defaults.
-/// If deny is set, will panic if schema has already been set.
-pub fn init_log_schema(config_paths: &[ConfigPath], deny_if_set: bool) -> Result<(), Vec<String>> {
-    let builder = load_builder_from_paths(config_paths)?;
-    vector_lib::config::init_log_schema(builder.global.log_schema, deny_if_set);
-    Ok(())
-}
-
 #[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq)]
 pub enum ConfigPath {
     File(PathBuf, FormatHint),

diff --git a/src/config/unit_test/mod.rs b/src/config/unit_test/mod.rs
@@ -83,11 +83,24 @@ impl UnitTest {
     }
 }
 
+/// Loads Log Schema from configurations and sets global schema.
+/// Once this is done, configurations can be correctly loaded using
+/// configured log schema defaults.
+/// If deny is set, will panic if schema has already been set.
+fn init_log_schema_from_paths(
+    config_paths: &[ConfigPath],
+    deny_if_set: bool,
+) -> Result<(), Vec<String>> {
+    let builder = config::loading::load_builder_from_paths(config_paths)?;
+    vector_lib::config::init_log_schema(builder.global.log_schema, deny_if_set);
+    Ok(())
+}
+
 pub async fn build_unit_tests_main(
     paths: &[ConfigPath],
     signal_handler: &mut signal::SignalHandler,
 ) -> Result<Vec<UnitTest>, Vec<String>> {
-    config::init_log_schema(paths, false)?;
+    init_log_schema_from_paths(paths, false)?;
     let mut secrets_backends_loader = loading::load_secret_backends_from_paths(paths)?;
     let config_builder = if secrets_backends_loader.has_secrets_to_retrieve() {
         let resolved_secrets = secrets_backends_loader

diff --git a/src/sources/docker_logs/mod.rs b/src/sources/docker_logs/mod.rs
@@ -70,8 +70,7 @@ pub struct DockerLogsConfig {
     /// By default, the [global `log_schema.host_key` option][global_host_key] is used.
     ///
     /// [global_host_key]: https://vector.dev/docs/reference/configuration/global-options/#log_schema.host_key
-    #[serde(default = "default_host_key")]
-    host_key: OptionalValuePath,
+    host_key: Option<OptionalValuePath>,
 
     /// Docker host to connect to.
     ///
@@ -171,7 +170,7 @@ pub struct DockerLogsConfig {
 impl Default for DockerLogsConfig {
     fn default() -> Self {
         Self {
-            host_key: default_host_key(),
+            host_key: None,
             docker_host: None,
             tls: None,
             exclude_containers: None,
@@ -187,10 +186,6 @@ impl Default for DockerLogsConfig {
     }
 }
 
-fn default_host_key() -> OptionalValuePath {
-    log_schema().host_key().cloned().into()
-}
-
 fn default_partial_event_marker_field() -> Option<String> {
     Some(event::PARTIAL.to_string())
 }
@@ -273,7 +268,12 @@ impl SourceConfig for DockerLogsConfig {
     }
 
     fn outputs(&self, global_log_namespace: LogNamespace) -> Vec<SourceOutput> {
-        let host_key = self.host_key.clone().path.map(LegacyKey::Overwrite);
+        let host_key = self
+            .host_key
+            .clone()
+            .unwrap_or(log_schema().host_key().cloned().into())
+            .path
+            .map(LegacyKey::Overwrite);
 
         let schema_definition = BytesDeserializerConfig
             .schema_definition(global_log_namespace.merge(self.log_namespace))
@@ -465,7 +465,10 @@ impl DockerLogsSource {
     ) -> crate::Result<DockerLogsSource> {
         let backoff_secs = config.retry_backoff_secs;
 
-        let host_key = config.host_key.clone();
+        let host_key = config
+            .host_key
+            .clone()
+            .unwrap_or(log_schema().host_key().cloned().into());
         let hostname = crate::get_hostname().ok();
 
         // Only logs created at, or after this moment are logged.

diff --git a/src/sources/file.rs b/src/sources/file.rs
@@ -122,9 +122,8 @@ pub struct FileConfig {
     /// Set to `""` to suppress this key.
     ///
     /// [global_host_key]: https://vector.dev/docs/reference/configuration/global-options/#log_schema.host_key
-    #[serde(default = "default_host_key")]
     #[configurable(metadata(docs::examples = "hostname"))]
-    pub host_key: OptionalValuePath,
+    pub host_key: Option<OptionalValuePath>,
 
     /// The directory used to persist file checkpoint positions.
     ///
@@ -253,10 +252,6 @@ fn default_file_key() -> OptionalValuePath {
     OptionalValuePath::from(owned_value_path!("file"))
 }
 
-fn default_host_key() -> OptionalValuePath {
-    log_schema().host_key().cloned().into()
-}
-
 const fn default_read_from() -> ReadFromConfig {
     ReadFromConfig::Beginning
 }
@@ -389,7 +384,7 @@ impl Default for FileConfig {
             max_line_bytes: default_max_line_bytes(),
             fingerprint: FingerprintConfig::default(),
             ignore_not_found: false,
-            host_key: default_host_key(),
+            host_key: None,
             offset_key: None,
             data_dir: None,
             glob_minimum_cooldown_ms: default_glob_minimum_cooldown_ms(),
@@ -453,7 +448,12 @@ impl SourceConfig for FileConfig {
 
     fn outputs(&self, global_log_namespace: LogNamespace) -> Vec<SourceOutput> {
         let file_key = self.file_key.clone().path.map(LegacyKey::Overwrite);
-        let host_key = self.host_key.clone().path.map(LegacyKey::Overwrite);
+        let host_key = self
+            .host_key
+            .clone()
+            .unwrap_or(log_schema().host_key().cloned().into())
+            .path
+            .map(LegacyKey::Overwrite);
 
         let offset_key = self
             .offset_key
@@ -561,7 +561,11 @@ pub fn file_source(
     };
 
     let event_metadata = EventMetadata {
-        host_key: config.host_key.clone().path,
+        host_key: config
+            .host_key
+            .clone()
+            .unwrap_or(log_schema().host_key().cloned().into())
+            .path,
         hostname: crate::get_hostname().ok(),
         file_key: config.file_key.clone().path,
         offset_key: config.offset_key.clone().and_then(|k| k.path),

diff --git a/src/sources/internal_logs.rs b/src/sources/internal_logs.rs
@@ -35,8 +35,7 @@ pub struct InternalLogsConfig {
     /// Set to `""` to suppress this key.
     ///
     /// [global_host_key]: https://vector.dev/docs/reference/configuration/global-options/#log_schema.host_key
-    #[serde(default = "default_host_key")]
-    host_key: OptionalValuePath,
+    host_key: Option<OptionalValuePath>,
 
     /// Overrides the name of the log field used to add the current process ID to each event.
     ///
@@ -52,10 +51,6 @@ pub struct InternalLogsConfig {
     log_namespace: Option<bool>,
 }
 
-fn default_host_key() -> OptionalValuePath {
-    log_schema().host_key().cloned().into()
-}
-
 fn default_pid_key() -> OptionalValuePath {
     OptionalValuePath::from(owned_value_path!("pid"))
 }
@@ -65,7 +60,7 @@ impl_generate_config_from_default!(InternalLogsConfig);
 impl Default for InternalLogsConfig {
     fn default() -> InternalLogsConfig {
         InternalLogsConfig {
-            host_key: default_host_key(),
+            host_key: None,
             pid_key: default_pid_key(),
             log_namespace: None,
         }
@@ -75,7 +70,12 @@ impl Default for InternalLogsConfig {
 impl InternalLogsConfig {
     /// Generates the `schema::Definition` for this component.
     fn schema_definition(&self, log_namespace: LogNamespace) -> Definition {
-        let host_key = self.host_key.clone().path.map(LegacyKey::Overwrite);
+        let host_key = self
+            .host_key
+            .clone()
+            .unwrap_or(log_schema().host_key().cloned().into())
+            .path
+            .map(LegacyKey::Overwrite);
         let pid_key = self.pid_key.clone().path.map(LegacyKey::Overwrite);
 
         // There is a global and per-source `log_namespace` config.
@@ -104,7 +104,11 @@ impl InternalLogsConfig {
 #[typetag::serde(name = "internal_logs")]
 impl SourceConfig for InternalLogsConfig {
     async fn build(&self, cx: SourceContext) -> crate::Result<super::Source> {
-        let host_key = self.host_key.clone().path;
+        let host_key = self
+            .host_key
+            .clone()
+            .unwrap_or(log_schema().host_key().cloned().into())
+            .path;
         let pid_key = self.pid_key.clone().path;
 
         let subscription = TraceSubscription::subscribe();

diff --git a/src/sources/internal_metrics.rs b/src/sources/internal_metrics.rs
@@ -54,7 +54,7 @@ impl Default for InternalMetricsConfig {
 
 /// Tag configuration for the `internal_metrics` source.
 #[configurable_component]
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Default)]
 #[serde(deny_unknown_fields, default)]
 pub struct TagsConfig {
     /// Overrides the name of the tag used to add the peer host to each metric.
@@ -66,8 +66,7 @@ pub struct TagsConfig {
     /// Set to `""` to suppress this key.
     ///
     /// [global_host_key]: https://vector.dev/docs/reference/configuration/global-options/#log_schema.host_key
-    #[serde(default = "default_host_key")]
-    pub host_key: OptionalValuePath,
+    pub host_key: Option<OptionalValuePath>,
 
     /// Sets the name of the tag to use to add the current process ID to each metric.
     ///
@@ -77,15 +76,6 @@ pub struct TagsConfig {
     pub pid_key: Option<String>,
 }
 
-impl Default for TagsConfig {
-    fn default() -> Self {
-        Self {
-            host_key: default_host_key(),
-            pid_key: None,
-        }
-    }
-}
-
 fn default_scrape_interval() -> Duration {
     Duration::from_secs_f64(1.0)
 }
@@ -94,10 +84,6 @@ fn default_namespace() -> String {
     "vector".to_owned()
 }
 
-fn default_host_key() -> OptionalValuePath {
-    log_schema().host_key().cloned().into()
-}
-
 impl_generate_config_from_default!(InternalMetricsConfig);
 
 #[async_trait::async_trait]
@@ -114,7 +100,11 @@ impl SourceConfig for InternalMetricsConfig {
         // namespace for created metrics is already "vector" by default.
         let namespace = self.namespace.clone();
 
-        let host_key = self.tags.host_key.clone();
+        let host_key = self
+            .tags
+            .host_key
+            .clone()
+            .unwrap_or(log_schema().host_key().cloned().into());
 
         let pid_key = self
             .tags
@@ -317,7 +307,7 @@ mod tests {
     async fn sets_tags() {
         let event = event_from_config(InternalMetricsConfig {
             tags: TagsConfig {
-                host_key: OptionalValuePath::new("my_host_key"),
+                host_key: Some(OptionalValuePath::new("my_host_key")),
                 pid_key: Some(String::from("my_pid_key")),
             },
             ..Default::default()

diff --git a/src/sources/socket/mod.rs b/src/sources/socket/mod.rs
@@ -211,7 +211,7 @@ impl SourceConfig for SocketConfig {
 
         let schema_definition = match &self.mode {
             Mode::Tcp(config) => {
-                let legacy_host_key = config.host_key().clone().path.map(LegacyKey::InsertIfEmpty);
+                let legacy_host_key = config.host_key().path.map(LegacyKey::InsertIfEmpty);
 
                 let legacy_port_key = config.port_key().clone().path.map(LegacyKey::InsertIfEmpty);
 
@@ -247,7 +247,7 @@ impl SourceConfig for SocketConfig {
                     )
             }
             Mode::Udp(config) => {
-                let legacy_host_key = config.host_key().clone().path.map(LegacyKey::InsertIfEmpty);
+                let legacy_host_key = config.host_key().path.map(LegacyKey::InsertIfEmpty);
 
                 let legacy_port_key = config.port_key().clone().path.map(LegacyKey::InsertIfEmpty);
 

diff --git a/src/sources/socket/tcp.rs b/src/sources/socket/tcp.rs
@@ -46,8 +46,7 @@ pub struct TcpConfig {
     /// Set to `""` to suppress this key.
     ///
     /// [global_host_key]: https://vector.dev/docs/reference/configuration/global-options/#log_schema.host_key
-    #[serde(default = "default_host_key")]
-    host_key: OptionalValuePath,
+    host_key: Option<OptionalValuePath>,
 
     /// Overrides the name of the log field used to add the peer host's port to each event.
     ///
@@ -106,7 +105,7 @@ impl TcpConfig {
             address,
             keepalive: None,
             shutdown_timeout_secs: default_shutdown_timeout_secs(),
-            host_key: default_host_key(),
+            host_key: None,
             port_key: default_port_key(),
             permit_origin: None,
             tls: None,
@@ -119,8 +118,8 @@ impl TcpConfig {
         }
     }
 
-    pub const fn host_key(&self) -> &OptionalValuePath {
-        &self.host_key
+    pub(super) fn host_key(&self) -> OptionalValuePath {
+        self.host_key.clone().unwrap_or(default_host_key())
     }
 
     pub const fn port_key(&self) -> &OptionalValuePath {
@@ -228,7 +227,12 @@ impl TcpSource for RawTcpSource {
                     now,
                 );
 
-                let legacy_host_key = self.config.host_key.clone().path;
+                let legacy_host_key = self
+                    .config
+                    .host_key
+                    .clone()
+                    .unwrap_or(default_host_key())
+                    .path;
 
                 self.log_namespace.insert_source_metadata(
                     SocketConfig::NAME,