chore: Enable pedantic Clippy lints in remaining sources and helpers (#26705)

Author: pront

clippy.toml

@@ -2,7 +2,7 @@ large-error-threshold = 256  # in bytes
 allow-unwrap-in-tests = true
 
 # Preserve product and component names in prose.
-doc-valid-idents = ["ClickHouse", "GreptimeCloud", "GreptimeDB", "InfluxDB", "JetStream", "NameNode", "OpenDAL", "OpenSearch", "PostgreSQL", "SQLx", "WebHDFS", ".."]
+doc-valid-idents = ["ClickHouse", "GreptimeCloud", "GreptimeDB", "InfluxDB", "JetStream", "MongoDB", "NameNode", "OpenDAL", "OpenSearch", "PostgreSQL", "SQLx", "StatsD", "WebHDFS", ".."]
 
 # for `disallowed_method`:
 # https://rust-lang.github.io/rust-clippy/master/index.html#disallowed_method

src/sources/kubernetes_logs/k8s_paths_provider.rs

@@ -128,7 +128,7 @@ fn extract_excluded_containers_for_pod(pod: &Pod) -> impl Iterator<Item = &str>
                 Some(value)
             })
             .flat_map(|containers| containers.split(','))
-            .map(|container| container.trim())
+            .map(str::trim)
     })
 }
 
@@ -188,7 +188,7 @@ fn real_glob(pattern: &str) -> impl Iterator<Item = PathBuf> + use<> {
         },
     )
     .expect("the pattern is supposed to always be correct")
-    .flat_map(|paths| paths.into_iter())
+    .flat_map(IntoIterator::into_iter)
 }
 
 fn filter_paths<'a>(
@@ -459,7 +459,7 @@ mod tests {
 
             let actual_paths: Vec<_> = list_pod_log_paths(mock_glob, &pod).collect();
             let expected_paths: Vec<_> = expected_paths.into_iter().map(PathBuf::from).collect();
-            assert_eq!(actual_paths, expected_paths)
+            assert_eq!(actual_paths, expected_paths);
         }
     }
 
@@ -532,7 +532,7 @@ mod tests {
                 actual_paths, expected_paths,
                 "failed for patterns {:?}",
                 &str_patterns
-            )
+            );
         }
     }
 
@@ -584,7 +584,7 @@ mod tests {
                 actual_paths, expected_paths,
                 "failed for patterns {:?}",
                 &str_patterns
-            )
+            );
         }
     }
 
@@ -626,7 +626,7 @@ mod tests {
                 actual_patterns, expected_patterns,
                 "failed for dir {:?} and containers {:?}",
                 &pod_logs_dir, &containers,
-            )
+            );
         }
     }
 }

src/sources/kubernetes_logs/mod.rs

@@ -1,7 +1,8 @@
+#![warn(clippy::pedantic)]
 //! This mod implements `kubernetes_logs` source.
 //! The scope of this source is to consume the log files that a kubelet keeps
 //! at "/var/log/pods" on the host of the Kubernetes Node when Vector itself is
-//! running inside the cluster as a DaemonSet.
+//! running inside the cluster as a `DaemonSet`.
 
 #![deny(missing_docs)]
 use std::{
@@ -83,6 +84,10 @@ use self::{
 /// The `self_node_name` value env var key.
 const SELF_NODE_NAME_ENV_KEY: &str = "VECTOR_SELF_NODE_NAME";
 
+#[allow(
+    clippy::struct_excessive_bools,
+    reason = "Restructuring these source configuration fields is deferred to preserve the existing API. https://github.com/vectordotdev/vector/issues/23659"
+)]
 /// Configuration for the `kubernetes_logs` source.
 #[serde_as]
 #[configurable_component(source("kubernetes_logs", "Collect Pod logs from Kubernetes Nodes."))]
@@ -318,11 +323,11 @@ impl GenerateConfig for Config {
 impl Default for Config {
     fn default() -> Self {
         Self {
-            extra_label_selector: "".to_string(),
-            extra_namespace_label_selector: "".to_string(),
+            extra_label_selector: String::new(),
+            extra_namespace_label_selector: String::new(),
             insert_namespace_fields: true,
             self_node_name: default_self_node_name_env_template(),
-            extra_field_selector: "".to_string(),
+            extra_field_selector: String::new(),
             auto_partial_merge: true,
             data_dir: None,
             pod_annotation_fields: pod_metadata_annotator::FieldsSpec::default(),
@@ -345,7 +350,7 @@ impl Default for Config {
             use_apiserver_cache: false,
             delay_deletion_ms: default_delay_deletion_ms(),
             log_namespace: None,
-            internal_metrics: Default::default(),
+            internal_metrics: FileInternalMetricsConfig::default(),
             rotate_wait: default_rotate_wait(),
         }
     }
@@ -370,6 +375,10 @@ impl SourceConfig for Config {
         ))
     }
 
+    #[allow(
+        clippy::too_many_lines,
+        reason = "Splitting this source routine is deferred to keep the lint cleanup mechanical. https://github.com/vectordotdev/vector/issues/23659"
+    )]
     fn outputs(&self, global_log_namespace: LogNamespace) -> Vec<SourceOutput> {
         let log_namespace = global_log_namespace.merge(self.log_namespace);
         let schema_definition = BytesDeserializerConfig
@@ -598,6 +607,10 @@ impl SourceConfig for Config {
     }
 }
 
+#[allow(
+    clippy::struct_excessive_bools,
+    reason = "Restructuring these source configuration fields is deferred to preserve the existing API. https://github.com/vectordotdev/vector/issues/23659"
+)]
 #[derive(Clone)]
 struct Source {
     client: Client,
@@ -723,6 +736,10 @@ impl Source {
         })
     }
 
+    #[allow(
+        clippy::too_many_lines,
+        reason = "Splitting this source routine is deferred to keep the lint cleanup mechanical. https://github.com/vectordotdev/vector/issues/23659"
+    )]
     async fn run(
         self,
         mut out: SourceSender,
@@ -1020,18 +1037,16 @@ impl Source {
                 shutdown,
                 Duration::from_secs(30), // more than enough time to propagate
             )
-            .map(|result| {
-                match result {
-                    Ok(Ok(())) => info!(message = "Event processing loop completed gracefully."),
-                    Ok(Err(_)) => emit!(StreamClosedError {
-                        count: events_count
-                    }),
-                    Err(error) => emit!(KubernetesLifecycleError {
-                        error,
-                        message: "Event processing loop timed out during the shutdown.",
-                        count: events_count,
-                    }),
-                };
+            .map(|result| match result {
+                Ok(Ok(())) => info!(message = "Event processing loop completed gracefully."),
+                Ok(Err(_)) => emit!(StreamClosedError {
+                    count: events_count
+                }),
+                Err(error) => emit!(KubernetesLifecycleError {
+                    error,
+                    message: "Event processing loop timed out during the shutdown.",
+                    count: events_count,
+                }),
             });
             slot.bind(Box::pin(fut));
         }
@@ -1131,11 +1146,11 @@ fn create_event(
         }
         // When LogNamespace::Legacy, only set when the `ingestion_timestamp_field` is configured.
         (LogNamespace::Legacy, Some(ingestion_timestamp_field)) => {
-            log.try_insert(ingestion_timestamp_field, Utc::now())
+            log.try_insert(ingestion_timestamp_field, Utc::now());
         }
         // The CRI/Docker parsers handle inserting the `log_schema().timestamp_key()` value.
         (LogNamespace::Legacy, None) => (),
-    };
+    }
 
     log.into()
 }
@@ -1183,15 +1198,15 @@ const fn default_max_line_bytes() -> usize {
 }
 
 const fn default_glob_minimum_cooldown_ms() -> Duration {
-    Duration::from_millis(60_000)
+    Duration::from_mins(1)
 }
 
 const fn default_fingerprint_lines() -> usize {
     1
 }
 
 const fn default_delay_deletion_ms() -> Duration {
-    Duration::from_millis(60_000)
+    Duration::from_mins(1)
 }
 
 const fn default_rotate_wait() -> Duration {
@@ -1236,6 +1251,10 @@ fn prepare_glob_patterns(paths: &[PathBuf], op: &str) -> crate::Result<Vec<glob:
 
 // This function constructs the effective field selector to use, based on
 // the specified configuration.
+#[allow(
+    clippy::unnecessary_wraps,
+    reason = "Removing this fallible API is deferred to keep source construction and error handling unchanged. https://github.com/vectordotdev/vector/issues/23659"
+)]
 fn prepare_field_selector(config: &Config, self_node_name: &str) -> crate::Result<String> {
     info!(
         message = "Obtained Kubernetes Node name to collect logs for (self).",
@@ -1252,6 +1271,10 @@ fn prepare_field_selector(config: &Config, self_node_name: &str) -> crate::Resul
 }
 
 // This function constructs the selector for a node to annotate entries with a node metadata.
+#[allow(
+    clippy::unnecessary_wraps,
+    reason = "Removing this fallible API is deferred to keep source construction and error handling unchanged. https://github.com/vectordotdev/vector/issues/23659"
+)]
 fn prepare_node_selector(self_node_name: &str) -> crate::Result<String> {
     Ok(format!("metadata.name={self_node_name}"))
 }
@@ -1372,9 +1395,9 @@ mod tests {
     #[test]
     fn test_config_serialization_insert_namespace_fields() {
         // Test that the flag serializes/deserializes correctly from YAML
-        let yaml_config = indoc! {r#"
+        let yaml_config = indoc! {r"
             insert_namespace_fields: false
-        "#};
+        "};
         let config: Config = serde_yaml::from_str(yaml_config).unwrap();
         assert_eq!(config.insert_namespace_fields, false);
 
@@ -1464,7 +1487,7 @@ mod tests {
             (
                 Config {
                     self_node_name: "qwe".to_owned(),
-                    extra_field_selector: "".to_owned(),
+                    extra_field_selector: String::new(),
                     ..Default::default()
                 },
                 "spec.nodeName=qwe",
@@ -1498,15 +1521,15 @@ mod tests {
             ),
             (
                 Config {
-                    extra_label_selector: "".to_owned(),
+                    extra_label_selector: String::new(),
                     ..Default::default()
                 }
                 .extra_label_selector,
                 "vector.dev/exclude!=true",
             ),
             (
                 Config {
-                    extra_namespace_label_selector: "".to_owned(),
+                    extra_namespace_label_selector: String::new(),
                     ..Default::default()
                 }
                 .extra_namespace_label_selector,
@@ -1536,6 +1559,10 @@ mod tests {
         }
     }
 
+    #[allow(
+        clippy::too_many_lines,
+        reason = "Splitting this fixture is deferred to keep the lint cleanup mechanical. https://github.com/vectordotdev/vector/issues/23659"
+    )]
     #[test]
     fn test_output_schema_definition_vector_namespace() {
         let definitions = serde_yaml::from_str::<Config>("")
@@ -1660,9 +1687,13 @@ mod tests {
                     )
                     .with_meaning(OwnedTargetPath::event_root(), "message")
             )
-        )
+        );
     }
 
+    #[allow(
+        clippy::too_many_lines,
+        reason = "Splitting this fixture is deferred to keep the lint cleanup mechanical. https://github.com/vectordotdev/vector/issues/23659"
+    )]
     #[test]
     fn test_output_schema_definition_legacy_namespace() {
         let definitions = serde_yaml::from_str::<Config>("")
@@ -1776,7 +1807,7 @@ mod tests {
                     None
                 )
             )
-        )
+        );
     }
 
     #[test]

src/sources/kubernetes_logs/namespace_metadata_annotator.rs

@@ -105,7 +105,7 @@ fn annotate_from_metadata(
     if let Some(labels) = &metadata.labels {
         let legacy_key_prefix = fields_spec.namespace_labels.path.as_ref().map(|k| &k.path);
 
-        for (key, value) in labels.iter() {
+        for (key, value) in labels {
             let key_path = path!(key);
             let legacy_key = legacy_key_prefix
                 .map(|k| k.concat(key_path))
@@ -117,7 +117,7 @@ fn annotate_from_metadata(
                 legacy_key,
                 path!("namespace_labels", key),
                 value.to_owned(),
-            )
+            );
         }
     }
 
@@ -128,7 +128,7 @@ fn annotate_from_metadata(
             .as_ref()
             .map(|k| &k.path);
 
-        for (key, value) in annotations.iter() {
+        for (key, value) in annotations {
             let key_path = path!(key);
             let legacy_key = legacy_key_prefix
                 .map(|k| k.concat(key_path))
@@ -140,7 +140,7 @@ fn annotate_from_metadata(
                 legacy_key,
                 path!("namespace_annotations", key),
                 value.to_owned(),
-            )
+            );
         }
     }
 }
@@ -152,6 +152,10 @@ mod tests {
 
     use super::*;
 
+    #[allow(
+        clippy::too_many_lines,
+        reason = "Splitting this fixture is deferred to keep the lint cleanup mechanical. https://github.com/vectordotdev/vector/issues/23659"
+    )]
     #[test]
     fn test_annotate_from_metadata() {
         let cases = vec![
@@ -541,7 +545,7 @@ mod tests {
             ),
         ];
 
-        for (fields_spec, metadata, expected, log_namespace) in cases.into_iter() {
+        for (fields_spec, metadata, expected, log_namespace) in cases {
             let mut log = LogEvent::default();
             annotate_from_metadata(&mut log, &fields_spec, &metadata, log_namespace);
             assert_eq!(log, expected);

src/sources/kubernetes_logs/node_metadata_annotator.rs

@@ -84,7 +84,7 @@ fn annotate_from_metadata(
     if let Some(labels) = &metadata.labels {
         let legacy_key_prefix = fields_spec.node_labels.path.as_ref().map(|k| &k.path);
 
-        for (key, value) in labels.iter() {
+        for (key, value) in labels {
             let key_path = path!(key);
             let legacy_key = legacy_key_prefix
                 .map(|k| k.concat(key_path))
@@ -96,7 +96,7 @@ fn annotate_from_metadata(
                 legacy_key,
                 path!("node_labels", key),
                 value.to_owned(),
-            )
+            );
         }
     }
 }
@@ -108,6 +108,10 @@ mod tests {
 
     use super::*;
 
+    #[allow(
+        clippy::too_many_lines,
+        reason = "Splitting this fixture is deferred to keep the lint cleanup mechanical. https://github.com/vectordotdev/vector/issues/23659"
+    )]
     #[test]
     fn test_annotate_from_metadata() {
         let cases = vec![
@@ -246,7 +250,7 @@ mod tests {
             ),
         ];
 
-        for (fields_spec, metadata, expected, log_namespace) in cases.into_iter() {
+        for (fields_spec, metadata, expected, log_namespace) in cases {
             let mut log = LogEvent::default();
             annotate_from_metadata(&mut log, &fields_spec, &metadata, log_namespace);
             assert_eq!(log, expected);