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]>
discord · Sep 23, 2024 · 144a814 · 144a814
1 parent 3ce836a
commit 144a814
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,