chore(dev): switch to Rust 1.85 (#22525)

* wip

* cargo clippy --fix

* more fixes

* last batch

* more updates

* no need to change msrv

* bump msrv - required for Option::expect

Cargo.toml

@@ -12,7 +12,7 @@ default-run = "vector"
 autobenches = false # our benchmarks are not runnable on their own either way
 # Minimum supported rust version
 # See docs/DEVELOPING.md for policy
-rust-version = "1.81"
+rust-version = "1.83"
 
 [[bin]]
 name = "vector"

lib/vector-config-common/src/schema/json_schema.rs

@@ -249,9 +249,7 @@ impl SchemaObject {
     /// and does not check any subschemas. Because of this, both `{}` and  `{"not": {}}` accept any type according
     /// to this method.
     pub fn has_type(&self, ty: InstanceType) -> bool {
-        self.instance_type
-            .as_ref()
-            .map_or(true, |x| x.contains(&ty))
+        self.instance_type.as_ref().is_none_or(|x| x.contains(&ty))
     }
 
     get_or_insert_default_fn!(metadata, Metadata);

lib/vector-config/src/schema/parser/query.rs

@@ -117,29 +117,25 @@ impl<'a> SchemaQueryBuilder<'a> {
                             let attr_matched = match self_attribute {
                                 CustomAttribute::Flag(key) => schema_attributes
                                     .get(key)
-                                    .map_or(false, |value| matches!(value, Value::Bool(true))),
+                                    .is_some_and(|value| matches!(value, Value::Bool(true))),
                                 CustomAttribute::KeyValue {
                                     key,
                                     value: attr_value,
                                 } => {
-                                    schema_attributes
-                                        .get(key)
-                                        .map_or(false, |value| match value {
-                                            // Check string values directly.
-                                            Value::String(schema_attr_value) => {
-                                                schema_attr_value == attr_value
-                                            }
-                                            // For arrays, try and convert each item to a string, and
-                                            // for the values that are strings, see if they match.
-                                            Value::Array(schema_attr_values) => {
-                                                schema_attr_values.iter().any(|value| {
-                                                    value
-                                                        .as_str()
-                                                        .map_or(false, |s| s == attr_value)
-                                                })
-                                            }
-                                            _ => false,
-                                        })
+                                    schema_attributes.get(key).is_some_and(|value| match value {
+                                        // Check string values directly.
+                                        Value::String(schema_attr_value) => {
+                                            schema_attr_value == attr_value
+                                        }
+                                        // For arrays, try and convert each item to a string, and
+                                        // for the values that are strings, see if they match.
+                                        Value::Array(schema_attr_values) => {
+                                            schema_attr_values.iter().any(|value| {
+                                                value.as_str().is_some_and(|s| s == attr_value)
+                                            })
+                                        }
+                                        _ => false,
+                                    })
                                 }
                             };
 

lib/vector-config/src/schema/visitors/unevaluated.rs

@@ -1,7 +1,4 @@
-use std::{
-    collections::{HashMap, HashSet},
-    convert::identity,
-};
+use std::collections::{HashMap, HashSet};
 
 use tracing::debug;
 use vector_config_common::schema::{
@@ -353,26 +350,23 @@ fn is_markable_schema(definitions: &Map<String, Schema>, schema: &SchemaObject)
         let has_object_subschema = subschemas
             .iter()
             .any(|schema| is_markable_schema(definitions, schema));
-        let has_referenced_object_subschema = subschemas
-            .iter()
-            .map(|subschema| {
-                subschema
-                    .reference
-                    .as_ref()
-                    .and_then(|reference| {
-                        let reference = get_cleaned_schema_reference(reference);
-                        definitions.get_key_value(reference)
-                    })
-                    .and_then(|(name, schema)| schema.as_object().map(|schema| (name, schema)))
-                    .map_or(false, |(name, schema)| {
-                        debug!(
-                            "Following schema reference '{}' for subschema markability.",
-                            name
-                        );
-                        is_markable_schema(definitions, schema)
-                    })
-            })
-            .any(identity);
+        let has_referenced_object_subschema = subschemas.iter().any(|subschema| {
+            subschema
+                .reference
+                .as_ref()
+                .and_then(|reference| {
+                    let reference = get_cleaned_schema_reference(reference);
+                    definitions.get_key_value(reference)
+                })
+                .and_then(|(name, schema)| schema.as_object().map(|schema| (name, schema)))
+                .is_some_and(|(name, schema)| {
+                    debug!(
+                        "Following schema reference '{}' for subschema markability.",
+                        name
+                    );
+                    is_markable_schema(definitions, schema)
+                })
+        });
 
         debug!(
             "Schema {} object subschema(s) and {} referenced subschemas.",

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

@@ -381,7 +381,7 @@ mod tests {
     }
 
     fn make_config<P: Debug>(name: &str, value: Option<P>) -> GlobalOptions {
-        toml::from_str(&value.map_or(String::new(), |value| format!(r#"{name} = {value:?}"#)))
+        toml::from_str(&value.map_or(String::new(), |value| format!(r"{name} = {value:?}")))
             .unwrap()
     }
 }

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

@@ -25,9 +25,9 @@ impl NoProxyInterceptor {
                 if scheme.is_some() && scheme != Some(expected_scheme) {
                     return false;
                 }
-                let matches = host.map_or(false, |host| {
+                let matches = host.is_some_and(|host| {
                     self.0.matches(host)
-                        || port.map_or(false, |port| {
+                        || port.is_some_and(|port| {
                             let url = format!("{host}:{port}");
                             self.0.matches(&url)
                         })

lib/vector-core/src/metrics/recorder.rs

@@ -46,9 +46,9 @@ impl Registry {
         let recency = recency.as_ref();
 
         for (key, counter) in self.registry.get_counter_handles() {
-            if recency.map_or(true, |recency| {
-                recency.should_store_counter(&key, &counter, &self.registry)
-            }) {
+            if recency
+                .is_none_or(|recency| recency.should_store_counter(&key, &counter, &self.registry))
+            {
                 // NOTE this will truncate if the value is greater than 2**52.
                 #[allow(clippy::cast_precision_loss)]
                 let value = counter.get_inner().load(Ordering::Relaxed) as f64;
@@ -57,16 +57,16 @@ impl Registry {
             }
         }
         for (key, gauge) in self.registry.get_gauge_handles() {
-            if recency.map_or(true, |recency| {
-                recency.should_store_gauge(&key, &gauge, &self.registry)
-            }) {
+            if recency
+                .is_none_or(|recency| recency.should_store_gauge(&key, &gauge, &self.registry))
+            {
                 let value = gauge.get_inner().load(Ordering::Relaxed);
                 let value = MetricValue::Gauge { value };
                 metrics.push(Metric::from_metric_kv(&key, value, timestamp));
             }
         }
         for (key, histogram) in self.registry.get_histogram_handles() {
-            if recency.map_or(true, |recency| {
+            if recency.is_none_or(|recency| {
                 recency.should_store_histogram(&key, &histogram, &self.registry)
             }) {
                 let value = histogram.get_inner().make_metric();

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
-channel = "1.83"
+channel = "1.85"
 profile = "default"

src/api/schema/metrics/source/file.rs

@@ -31,7 +31,7 @@ impl<'a> FileSourceMetricFile<'a> {
 }
 
 #[Object]
-impl<'a> FileSourceMetricFile<'a> {
+impl FileSourceMetricFile<'_> {
     /// File name
     async fn name(&self) -> &str {
         &*self.name

src/encoding_transcode.rs

@@ -84,10 +84,7 @@ impl Decoder {
         // processing, we handle it centrally here. Also, the BOM does not serve
         // any more use for us, since the source encoding is already pre-identified
         // as part of decoder initialization.
-        if output
-            .get(..BOM_UTF8_LEN)
-            .map_or(false, |start| start == BOM_UTF8)
-        {
+        if output.get(..BOM_UTF8_LEN) == Some(BOM_UTF8) {
             emit!(DecoderBomRemoval {
                 from_encoding: self.inner.encoding().name()
             });

src/secrets/exec.rs

@@ -117,12 +117,10 @@ async fn query_backend(
     let mut stderr_stream = child
         .stderr
         .map(|s| codec::FramedRead::new(s, codec::LinesCodec::new()))
-        .take()
         .ok_or("unable to acquire stderr")?;
     let mut stdout_stream = child
         .stdout
         .map(|s| codec::FramedRead::new(s, codec::BytesCodec::new()))
-        .take()
         .ok_or("unable to acquire stdout")?;
 
     let query = serde_json::to_vec(&query)?;

src/sinks/aws_cloudwatch_metrics/mod.rs

@@ -244,7 +244,6 @@ impl CloudWatchMetricsSvc {
                     normalizer.normalize(event.into_metric()).map(|mut metric| {
                         let namespace = metric
                             .take_namespace()
-                            .take()
                             .unwrap_or_else(|| default_namespace.clone());
                         Ok(EncodedEvent::new(
                             PartitionInnerBuffer::new(metric, namespace),

src/sinks/azure_blob/integration_tests.rs

@@ -55,7 +55,7 @@ async fn azure_blob_healthcheck_unknown_container() {
     };
     let client = azure_common::config::build_client(
         config.connection_string.map(Into::into),
-        config.storage_account.map(Into::into),
+        config.storage_account,
         config.container_name.clone(),
         config.endpoint.clone(),
     )
@@ -245,7 +245,7 @@ impl AzureBlobSinkConfig {
     fn to_sink(&self) -> VectorSink {
         let client = azure_common::config::build_client(
             self.connection_string.clone().map(Into::into),
-            self.storage_account.clone().map(Into::into),
+            self.storage_account.clone(),
             self.container_name.clone(),
             self.endpoint.clone(),
         )
@@ -264,7 +264,7 @@ impl AzureBlobSinkConfig {
     pub async fn list_blobs(&self, prefix: String) -> Vec<String> {
         let client = azure_common::config::build_client(
             self.connection_string.clone().map(Into::into),
-            self.storage_account.clone().map(Into::into),
+            self.storage_account.clone(),
             self.container_name.clone(),
             self.endpoint.clone(),
         )
@@ -293,7 +293,7 @@ impl AzureBlobSinkConfig {
     pub async fn get_blob(&self, blob: String) -> (Blob, Vec<String>) {
         let client = azure_common::config::build_client(
             self.connection_string.clone().map(Into::into),
-            self.storage_account.clone().map(Into::into),
+            self.storage_account.clone(),
             self.container_name.clone(),
             self.endpoint.clone(),
         )
@@ -330,7 +330,7 @@ impl AzureBlobSinkConfig {
     async fn ensure_container(&self) {
         let client = azure_common::config::build_client(
             self.connection_string.clone().map(Into::into),
-            self.storage_account.clone().map(Into::into),
+            self.storage_account.clone(),
             self.container_name.clone(),
             self.endpoint.clone(),
         )

src/sinks/datadog/traces/sink.rs

@@ -61,10 +61,10 @@ impl Partitioner for EventPartitioner {
                     .map(|s| s.to_string_lossy().into_owned()),
                 target_tps: t
                     .get(event_path!("target_tps"))
-                    .and_then(|tps| tps.as_integer().map(Into::into)),
+                    .and_then(|tps| tps.as_integer()),
                 error_tps: t
                     .get(event_path!("error_tps"))
-                    .and_then(|tps| tps.as_integer().map(Into::into)),
+                    .and_then(|tps| tps.as_integer()),
             },
         }
     }

src/sinks/vector/mod.rs

@@ -222,7 +222,6 @@ mod tests {
         .await
         .into_iter()
         .flatten()
-        .map(Into::into)
         .collect()
     }
 

src/sources/gcp_pubsub.rs

@@ -717,7 +717,7 @@ fn is_reset(error: &Status) -> bool {
         .and_then(|source| source.downcast_ref::<hyper::Error>())
         .and_then(|error| error.source())
         .and_then(|source| source.downcast_ref::<h2::Error>())
-        .map_or(false, |error| error.is_remote() && error.is_reset())
+        .is_some_and(|error| error.is_remote() && error.is_reset())
 }
 
 #[pin_project::pin_project]

src/sources/host_metrics/cgroups.rs

@@ -416,11 +416,11 @@ define_stat_struct! { MemoryStat(
 )}
 
 fn is_dir(path: impl AsRef<Path>) -> bool {
-    std::fs::metadata(path.as_ref()).map_or(false, |metadata| metadata.is_dir())
+    std::fs::metadata(path.as_ref()).is_ok_and(|metadata| metadata.is_dir())
 }
 
 fn is_file(path: impl AsRef<Path>) -> bool {
-    std::fs::metadata(path.as_ref()).map_or(false, |metadata| metadata.is_file())
+    std::fs::metadata(path.as_ref()).is_ok_and(|metadata| metadata.is_file())
 }
 
 /// Join a base directory path with a cgroup name.

src/sources/internal_metrics.rs

@@ -145,7 +145,7 @@ struct InternalMetrics<'a> {
     shutdown: ShutdownSignal,
 }
 
-impl<'a> InternalMetrics<'a> {
+impl InternalMetrics<'_> {
     async fn run(mut self) -> Result<(), ()> {
         let events_received = register!(EventsReceived);
         let bytes_received = register!(BytesReceived::from(Protocol::INTERNAL));

src/sources/prometheus/parser.rs

@@ -103,7 +103,7 @@ fn reparse_groups(
                     }
                     let drop_last = buckets
                         .last()
-                        .map_or(false, |bucket| bucket.bucket == f64::INFINITY);
+                        .is_some_and(|bucket| bucket.bucket == f64::INFINITY);
                     if drop_last {
                         buckets.pop();
                     }

src/sources/splunk_hec/mod.rs

@@ -1445,7 +1445,7 @@ mod tests {
         b.body(message.to_owned())
     }
 
-    async fn send_with<'a>(
+    async fn send_with(
         address: SocketAddr,
         api: &str,
         message: &str,
@@ -1456,7 +1456,7 @@ mod tests {
         b.send().await.unwrap().status().as_u16()
     }
 
-    async fn send_with_response<'a>(
+    async fn send_with_response(
         address: SocketAddr,
         api: &str,
         message: &str,

src/template.rs

@@ -348,7 +348,7 @@ fn parse_template(src: &str) -> Result<Vec<Part>, TemplateParseError> {
 fn render_metric_field<'a>(key: &str, metric: &'a Metric) -> Option<&'a str> {
     match key {
         "name" => Some(metric.name()),
-        "namespace" => metric.namespace().map(Into::into),
+        "namespace" => metric.namespace(),
         _ if key.starts_with("tags.") => metric.tags().and_then(|tags| tags.get(&key[5..])),
         _ => None,
     }

src/transforms/dedupe/config.rs

@@ -93,7 +93,7 @@ mod tests {
         crate::test_util::test_generate_config::<DedupeConfig>();
     }
 
-    fn make_match_transform_config(
+    const fn make_match_transform_config(
         num_events: usize,
         fields: Vec<ConfigTargetPath>,
     ) -> DedupeConfig {

src/transforms/dedupe/mod.rs

@@ -21,7 +21,7 @@ pub mod common {
         pub num_events: NonZeroUsize,
     }
 
-    pub fn default_cache_config() -> CacheConfig {
+    pub const fn default_cache_config() -> CacheConfig {
         CacheConfig {
             num_events: NonZeroUsize::new(5000).expect("static non-zero number"),
         }

vdev/src/commands/e2e/mod.rs

@@ -1,9 +1,9 @@
 crate::cli_subcommands! {
-    r#"Manage end-to-end test environments...
+    r"Manage end-to-end test environments...
 
 These test setups are organized into a set of integrations, located in subdirectories
 `scripts/e2e`.  For each integration, there is a matrix of environments, described in the
-`matrix` setting in the `test.yaml` file contained therein."#
+`matrix` setting in the `test.yaml` file contained therein."
 
     mod show,
     mod start,

vdev/src/commands/integration/mod.rs

@@ -1,9 +1,9 @@
 crate::cli_subcommands! {
-    r#"Manage integration test environments...
+    r"Manage integration test environments...
 
 These test setups are organized into a set of integrations, located in subdirectories
 `scripts/integration`.  For each integration, there is a matrix of environments, described in the
-`matrix` setting in the `test.yaml` file contained therein."#
+`matrix` setting in the `test.yaml` file contained therein."
 
     mod show,
     mod start,