chore(deps): Bump Rust version to 1.81.0

Tiltfile

@@ -7,7 +7,7 @@ load('ext://helm_resource', 'helm_resource', 'helm_repo')
 docker_build(
     ref='timberio/vector',
     context='.',
-    build_args={'RUST_VERSION': '1.80.0'},
+    build_args={'RUST_VERSION': '1.81.0'},
     dockerfile='tilt/Dockerfile'
     )
 

lib/prometheus-parser/src/line.rs

@@ -373,7 +373,7 @@ fn parse_name(input: &str) -> IResult<String> {
 }
 
 fn trim_space(input: &str) -> &str {
-    input.trim_start_matches(|c| c == ' ' || c == '\t')
+    input.trim_start_matches([' ', '\t'])
 }
 
 fn sp<'a, E: ParseError<&'a str>>(i: &'a str) -> nom::IResult<&'a str, &'a str, E> {

lib/vector-core/src/event/log_event.rs

@@ -625,7 +625,7 @@ impl EventDataEq for LogEvent {
 
 #[cfg(any(test, feature = "test"))]
 mod test_utils {
-    use super::*;
+    use super::{log_schema, Bytes, LogEvent, Utc};
 
     // these rely on the global log schema, which is no longer supported when using the
     // "LogNamespace::Vector" namespace.

lib/vector-core/src/schema/definition.rs

@@ -537,7 +537,7 @@ impl Definition {
 
 #[cfg(any(test, feature = "test"))]
 mod test_utils {
-    use super::*;
+    use super::{Definition, Kind};
     use crate::event::{Event, LogEvent};
 
     impl Definition {

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
-channel = "1.80"
+channel = "1.81"
 profile = "default"

src/api/schema/metrics/sink/generic.rs

@@ -9,7 +9,7 @@ use crate::{
 pub struct GenericSinkMetrics(Vec<Metric>);
 
 impl GenericSinkMetrics {
-    pub fn new(metrics: Vec<Metric>) -> Self {
+    pub const fn new(metrics: Vec<Metric>) -> Self {
         Self(metrics)
     }
 }

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

@@ -57,7 +57,7 @@ impl<'a> FileSourceMetricFile<'a> {
 pub struct FileSourceMetrics(Vec<Metric>);
 
 impl FileSourceMetrics {
-    pub fn new(metrics: Vec<Metric>) -> Self {
+    pub const fn new(metrics: Vec<Metric>) -> Self {
         Self(metrics)
     }
 

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

@@ -9,7 +9,7 @@ use crate::{
 pub struct GenericSourceMetrics(Vec<Metric>);
 
 impl GenericSourceMetrics {
-    pub fn new(metrics: Vec<Metric>) -> Self {
+    pub const fn new(metrics: Vec<Metric>) -> Self {
         Self(metrics)
     }
 }

src/api/schema/metrics/transform/generic.rs

@@ -9,7 +9,7 @@ use crate::{
 pub struct GenericTransformMetrics(Vec<Metric>);
 
 impl GenericTransformMetrics {
-    pub fn new(metrics: Vec<Metric>) -> Self {
+    pub const fn new(metrics: Vec<Metric>) -> Self {
         Self(metrics)
     }
 }

src/api/server.rs

@@ -45,12 +45,11 @@ impl Server {
         let _guard = handle.enter();
 
         let addr = config.api.address.expect("No socket address");
-        let incoming = AddrIncoming::bind(&addr).map_err(|error| {
+        let incoming = AddrIncoming::bind(&addr).inspect_err(|error| {
             emit!(SocketBindError {
                 mode: SocketMode::Tcp,
-                error: &error,
+                error,
             });
-            error
         })?;
 
         let span = Span::current();

src/app.rs

@@ -434,12 +434,11 @@ impl FinishedApplication {
 fn get_log_levels(default: &str) -> String {
     std::env::var("VECTOR_LOG")
         .or_else(|_| {
-            std::env::var("LOG").map(|log| {
+            std::env::var("LOG").inspect(|_log| {
                 warn!(
                     message =
                         "DEPRECATED: Use of $LOG is deprecated. Please use $VECTOR_LOG instead."
                 );
-                log
             })
         })
         .unwrap_or_else(|_| default.into())

src/aws/mod.rs

@@ -363,7 +363,7 @@ struct MeasuredBody {
 }
 
 impl MeasuredBody {
-    fn new(body: SdkBody, shared_bytes_sent: Arc<AtomicUsize>) -> Self {
+    const fn new(body: SdkBody, shared_bytes_sent: Arc<AtomicUsize>) -> Self {
         Self {
             inner: body,
             shared_bytes_sent,

src/components/validation/mod.rs

@@ -132,7 +132,7 @@ pub struct ValidationConfiguration {
 
 impl ValidationConfiguration {
     /// Creates a new `ValidationConfiguration` for a source.
-    pub fn from_source(
+    pub const fn from_source(
         component_name: &'static str,
         log_namespace: LogNamespace,
         component_configurations: Vec<ComponentTestCaseConfig>,
@@ -146,7 +146,7 @@ impl ValidationConfiguration {
     }
 
     /// Creates a new `ValidationConfiguration` for a transform.
-    pub fn from_transform(
+    pub const fn from_transform(
         component_name: &'static str,
         log_namespace: LogNamespace,
         component_configurations: Vec<ComponentTestCaseConfig>,
@@ -160,7 +160,7 @@ impl ValidationConfiguration {
     }
 
     /// Creates a new `ValidationConfiguration` for a sink.
-    pub fn from_sink(
+    pub const fn from_sink(
         component_name: &'static str,
         log_namespace: LogNamespace,
         component_configurations: Vec<ComponentTestCaseConfig>,

src/generate.rs

@@ -119,7 +119,7 @@ pub(crate) fn generate_example(
 ) -> Result<String, Vec<String>> {
     let components: Vec<Vec<_>> = opts
         .expression
-        .split(|c| c == '|' || c == '/')
+        .split(['|', '/'])
         .map(|s| {
             s.split(',')
                 .map(|s| s.trim().to_string())

src/http.rs

@@ -139,13 +139,9 @@ where
 
             // Handle the errors and extract the response.
             let response = response_result
-                .map_err(|error| {
+                .inspect_err(|error| {
                     // Emit the error into the internal events system.
-                    emit!(http_client::GotHttpWarning {
-                        error: &error,
-                        roundtrip
-                    });
-                    error
+                    emit!(http_client::GotHttpWarning { error, roundtrip });
                 })
                 .context(CallRequestSnafu)?;
 

src/sinks/amqp/service.rs

@@ -22,7 +22,7 @@ pub(super) struct AmqpRequest {
 }
 
 impl AmqpRequest {
-    pub(super) fn new(
+    pub(super) const fn new(
         body: Bytes,
         exchange: String,
         routing_key: String,

src/sinks/aws_kinesis/firehose/integration_tests.rs

@@ -101,7 +101,7 @@ async fn firehose_put_records_without_partition_key() {
         .expect("Could not build HTTP client");
 
     let response = client
-        .get(&format!("{}/{}/_search", common.base_url, stream))
+        .get(format!("{}/{}/_search", common.base_url, stream))
         .json(&json!({
             "query": { "query_string": { "query": "*" } }
         }))
@@ -213,7 +213,7 @@ async fn firehose_put_records_with_partition_key() {
         .expect("Could not build HTTP client");
 
     let response = client
-        .get(&format!("{}/{}/_search", common.base_url, stream))
+        .get(format!("{}/{}/_search", common.base_url, stream))
         .json(&json!({
             "query": { "query_string": { "query": "*" } }
         }))

src/sinks/aws_kinesis/streams/integration_tests.rs

@@ -76,9 +76,8 @@ async fn kinesis_put_records_with_partition_key() {
 
     let mut output_lines = records
         .into_iter()
-        .map(|e| {
+        .inspect(|e| {
             assert_eq!(partition_value, e.partition_key());
-            e
         })
         .map(|e| String::from_utf8(e.data.into_inner()).unwrap())
         .collect::<Vec<_>>();

src/sinks/aws_s_s/sink.rs

@@ -2,15 +2,6 @@ use super::{client::Client, request_builder::SSRequestBuilder, service::SSServic
 use crate::sinks::aws_s_s::retry::SSRetryLogic;
 use crate::sinks::prelude::*;
 
-#[derive(Clone, Copy, Debug, Default)]
-pub(crate) struct SqsSinkDefaultBatchSettings;
-
-impl SinkBatchSettings for SqsSinkDefaultBatchSettings {
-    const MAX_EVENTS: Option<usize> = Some(1);
-    const MAX_BYTES: Option<usize> = Some(262_144);
-    const TIMEOUT_SECS: f64 = 1.0;
-}
-
 #[derive(Clone)]
 pub(super) struct SSSink<C, E>
 where

src/sinks/azure_common/service.rs

@@ -17,7 +17,7 @@ pub struct AzureBlobService {
 }
 
 impl AzureBlobService {
-    pub fn new(client: Arc<ContainerClient>) -> AzureBlobService {
+    pub const fn new(client: Arc<ContainerClient>) -> AzureBlobService {
         AzureBlobService { client }
     }
 }

src/sinks/datadog/logs/sink.rs

@@ -39,7 +39,7 @@ pub struct LogSinkBuilder<S> {
 }
 
 impl<S> LogSinkBuilder<S> {
-    pub fn new(
+    pub const fn new(
         transformer: Transformer,
         service: S,
         default_api_key: Arc<str>,

src/sinks/datadog/traces/request_builder.rs

@@ -71,7 +71,7 @@ pub struct DatadogTracesRequestBuilder {
 }
 
 impl DatadogTracesRequestBuilder {
-    pub fn new(
+    pub const fn new(
         api_key: Arc<str>,
         endpoint_configuration: DatadogTracesEndpointConfiguration,
         compression: Compression,

src/sinks/elasticsearch/encoder.rs

@@ -125,12 +125,12 @@ impl Encoder<Vec<ProcessedEvent>> for ElasticsearchEncoder {
             )?;
             written_bytes +=
                 as_tracked_write::<_, _, io::Error>(writer, &log, |mut writer, log| {
-                    writer.write_all(&[b'\n'])?;
+                    writer.write_all(b"\n")?;
                     // False positive clippy hit on the following line. Clippy wants us to skip the
                     // borrow, but then the value is moved for the following line.
                     #[allow(clippy::needless_borrows_for_generic_args)]
                     serde_json::to_writer(&mut writer, log)?;
-                    writer.write_all(&[b'\n'])?;
+                    writer.write_all(b"\n")?;
                     Ok(())
                 })?;
         }

src/sinks/elasticsearch/integration_tests.rs

@@ -199,7 +199,7 @@ async fn structures_events_correctly() {
     flush(common).await.unwrap();
 
     let response = reqwest::Client::new()
-        .get(&format!("{}/{}/_search", base_url, index))
+        .get(format!("{}/{}/_search", base_url, index))
         .json(&json!({
             "query": { "query_string": { "query": "*" } }
         }))
@@ -669,7 +669,7 @@ async fn run_insert_tests_with_config(
 
     let client = create_http_client();
     let mut response = client
-        .get(&format!("{}/{}/_search", base_url, index))
+        .get(format!("{}/{}/_search", base_url, index))
         .basic_auth("elastic", Some("vector"))
         .json(&json!({
             "query": { "query_string": { "query": "*" } }
@@ -758,7 +758,7 @@ async fn run_insert_tests_with_multiple_endpoints(config: &ElasticsearchConfig)
     let mut total = 0;
     for base_url in base_urls {
         if let Ok(response) = client
-            .get(&format!("{}/{}/_search", base_url, index))
+            .get(format!("{}/{}/_search", base_url, index))
             .basic_auth("elastic", Some("vector"))
             .json(&json!({
                 "query": { "query_string": { "query": "*" } }

src/sinks/file/bytes_path.rs

@@ -14,7 +14,7 @@ pub struct BytesPath {
 
 impl BytesPath {
     #[cfg(unix)]
-    pub fn new(path: Bytes) -> Self {
+    pub const fn new(path: Bytes) -> Self {
         Self { path }
     }
     #[cfg(windows)]

src/sinks/greptimedb/logs/integration_tests.rs

@@ -74,7 +74,7 @@ impl GreptimeClient {
 
     async fn create_pipeline(&self, pipeline_name: &str, pipeline_content: &str) {
         self.client
-            .post(&format!(
+            .post(format!(
                 "{}/v1/events/pipelines/{}",
                 self.endpoint, pipeline_name
             ))
@@ -87,7 +87,7 @@ impl GreptimeClient {
 
     async fn query(&self, sql: &str) -> String {
         self.client
-            .get(&format!("{}/v1/sql", self.endpoint))
+            .get(format!("{}/v1/sql", self.endpoint))
             .query(&[("sql", sql)])
             .send()
             .await

src/sinks/greptimedb/metrics/integration_tests.rs

@@ -29,7 +29,7 @@ async fn test_greptimedb_sink() {
 
     // Drop the table and data inside
     let _ = query_client
-        .get(&format!(
+        .get(format!(
             "{}/v1/sql",
             std::env::var("GREPTIMEDB_HTTP").unwrap_or_else(|_| "http://localhost:4000".to_owned())
         ))
@@ -45,7 +45,7 @@ async fn test_greptimedb_sink() {
     run_and_assert_sink_compliance(sink, stream::iter(events), &SINK_TAGS).await;
 
     let query_response = query_client
-        .get(&format!(
+        .get(format!(
             "{}/v1/sql",
             std::env::var("GREPTIMEDB_HTTP").unwrap_or_else(|_| "http://localhost:4000".to_owned())
         ))
@@ -85,7 +85,7 @@ new_naming = true
 
     // Drop the table and data inside
     let _ = query_client
-        .get(&format!(
+        .get(format!(
             "{}/v1/sql",
             std::env::var("GREPTIMEDB_HTTP").unwrap_or_else(|_| "http://localhost:4000".to_owned())
         ))
@@ -101,7 +101,7 @@ new_naming = true
     run_and_assert_sink_compliance(sink, stream::iter(events), &SINK_TAGS).await;
 
     let query_response = query_client
-        .get(&format!(
+        .get(format!(
             "{}/v1/sql",
             std::env::var("GREPTIMEDB_HTTP").unwrap_or_else(|_| "http://localhost:4000".to_owned())
         ))

src/sinks/influxdb/mod.rs

@@ -471,7 +471,7 @@ pub mod test_util {
 
     pub(crate) async fn query_v1(endpoint: &str, query: &str) -> reqwest::Response {
         client()
-            .get(&format!("{}/query", endpoint))
+            .get(format!("{}/query", endpoint))
             .query(&[("q", query)])
             .send()
             .await

src/sinks/new_relic/model.rs

@@ -45,7 +45,7 @@ pub(super) struct MetricData {
 }
 
 impl MetricsApiModel {
-    pub(super) fn new(metrics: Vec<MetricData>) -> Self {
+    pub(super) const fn new(metrics: Vec<MetricData>) -> Self {
         Self([MetricDataStore { metrics }])
     }
 }
@@ -148,7 +148,7 @@ impl TryFrom<Vec<Event>> for MetricsApiModel {
 pub(super) struct EventsApiModel(pub Vec<ObjectMap>);
 
 impl EventsApiModel {
-    pub(super) fn new(events_array: Vec<ObjectMap>) -> Self {
+    pub(super) const fn new(events_array: Vec<ObjectMap>) -> Self {
         Self(events_array)
     }
 }
@@ -265,7 +265,7 @@ pub(super) enum Timestamp {
 }
 
 impl LogsApiModel {
-    pub(super) fn new(logs: Vec<LogMessage>) -> Self {
+    pub(super) const fn new(logs: Vec<LogMessage>) -> Self {
         Self([LogDataStore { logs }])
     }
 }

src/sinks/prometheus/collector.rs

@@ -301,7 +301,7 @@ impl StringCollector {
         let mut result = String::with_capacity(key.len() + value.len() + 3);
         result.push_str(key);
         result.push_str("=\"");
-        while let Some(i) = value.find(|ch| ch == '\\' || ch == '"') {
+        while let Some(i) = value.find(['\\', '"']) {
             result.push_str(&value[..i]);
             result.push('\\');
             // Ugly but works because we know the character at `i` is ASCII