chore(dev): upgrade to Rust 2024 edition

Cargo.toml

@@ -2,7 +2,7 @@
 name = "vector"
 version = "0.49.0"
 authors = ["Vector Contributors <vector@datadoghq.com>"]
-edition = "2021"
+edition = "2024"
 description = "A lightweight and ultra-fast tool for building observability pipelines"
 homepage = "https://vector.dev"
 license = "MPL-2.0"

benches/batch.rs

@@ -1,13 +1,13 @@
 use std::{convert::Infallible, time::Duration};
 
 use bytes::{BufMut, Bytes, BytesMut};
-use criterion::{criterion_group, Criterion, SamplingMode, Throughput};
-use futures::{future, stream, SinkExt, StreamExt};
+use criterion::{Criterion, SamplingMode, Throughput, criterion_group};
+use futures::{SinkExt, StreamExt, future, stream};
 use vector::{
     sinks::util::{
-        batch::{Batch, BatchConfig, BatchError, BatchSettings, BatchSize, PushResult},
         BatchSink, Buffer, Compression, EncodedEvent, Merged, Partition, PartitionBatchSink,
         SinkBatchSettings,
+        batch::{Batch, BatchConfig, BatchError, BatchSettings, BatchSize, PushResult},
     },
     test_util::{random_lines, runtime},
 };

benches/codecs/character_delimited_bytes.rs

@@ -2,13 +2,13 @@ use std::{fmt, time::Duration};
 
 use bytes::BytesMut;
 use criterion::{
-    criterion_group, measurement::WallTime, BatchSize, BenchmarkGroup, BenchmarkId, Criterion,
-    SamplingMode, Throughput,
+    BatchSize, BenchmarkGroup, BenchmarkId, Criterion, SamplingMode, Throughput, criterion_group,
+    measurement::WallTime,
 };
 use tokio_util::codec::Decoder;
 use vector_lib::codecs::{
-    decoding::{Deserializer, Framer},
     BytesDeserializer, CharacterDelimitedDecoder,
+    decoding::{Deserializer, Framer},
 };
 
 #[derive(Debug)]

benches/codecs/encoder.rs

@@ -2,12 +2,12 @@ use std::time::Duration;
 
 use bytes::{BufMut, BytesMut};
 use criterion::{
-    criterion_group, measurement::WallTime, BatchSize, BenchmarkGroup, Criterion, SamplingMode,
-    Throughput,
+    BatchSize, BenchmarkGroup, Criterion, SamplingMode, Throughput, criterion_group,
+    measurement::WallTime,
 };
 use tokio_util::codec::Encoder;
 use vector::event::{Event, LogEvent};
-use vector_lib::codecs::{encoding::Framer, JsonSerializerConfig, NewlineDelimitedEncoder};
+use vector_lib::codecs::{JsonSerializerConfig, NewlineDelimitedEncoder, encoding::Framer};
 use vector_lib::{btreemap, byte_size_of::ByteSizeOf};
 
 #[derive(Debug, Clone)]

benches/codecs/newline_bytes.rs

@@ -2,12 +2,12 @@ use std::{fmt, time::Duration};
 
 use bytes::BytesMut;
 use criterion::{
-    criterion_group, measurement::WallTime, BatchSize, BenchmarkGroup, BenchmarkId, Criterion,
-    SamplingMode, Throughput,
+    BatchSize, BenchmarkGroup, BenchmarkId, Criterion, SamplingMode, Throughput, criterion_group,
+    measurement::WallTime,
 };
 use tokio_util::codec::Decoder;
 use vector_lib::codecs::{
-    decoding::Deserializer, decoding::Framer, BytesDeserializer, NewlineDelimitedDecoder,
+    BytesDeserializer, NewlineDelimitedDecoder, decoding::Deserializer, decoding::Framer,
 };
 
 #[derive(Debug)]

benches/distribution_statistic.rs

@@ -1,4 +1,4 @@
-use criterion::{criterion_group, criterion_main, BatchSize, Criterion};
+use criterion::{BatchSize, Criterion, criterion_group, criterion_main};
 use rand::distr::Distribution;
 use rand::{distr::Uniform, seq::SliceRandom};
 use vector::{event::metric::Sample, sinks::util::statistic::DistributionStatistic};

benches/dnstap/mod.rs

@@ -1,7 +1,7 @@
-use base64::engine::general_purpose::STANDARD;
 use base64::Engine;
+use base64::engine::general_purpose::STANDARD;
 use bytes::Bytes;
-use criterion::{criterion_group, criterion_main, BatchSize, Criterion, Throughput};
+use criterion::{BatchSize, Criterion, Throughput, criterion_group, criterion_main};
 use dnsmsg_parser::dns_message_parser::DnsParserOptions;
 use dnstap_parser::parser::DnstapParser;
 use vector::event::LogEvent;

benches/enrichment_tables.rs

@@ -1,13 +1,13 @@
 use std::time::SystemTime;
 
 use chrono::prelude::*;
-use criterion::{criterion_group, criterion_main, BatchSize, Criterion};
+use criterion::{BatchSize, Criterion, criterion_group, criterion_main};
 use vector::enrichment_tables::file::FileData;
 use vector::enrichment_tables::{
+    Condition, Table,
     file::File,
     geoip::{Geoip, GeoipConfig},
     mmdb::{Mmdb, MmdbConfig},
-    Condition, Table,
 };
 use vector_lib::enrichment::Case;
 use vrl::value::{ObjectMap, Value};

benches/event.rs

@@ -1,5 +1,5 @@
 use bytes::Bytes;
-use criterion::{criterion_group, BatchSize, Criterion};
+use criterion::{BatchSize, Criterion, criterion_group};
 use vector::event::LogEvent;
 use vrl::event_path;
 

benches/files.rs

@@ -1,16 +1,16 @@
 use std::{convert::TryInto, path::PathBuf, time::Duration};
 
 use bytes::Bytes;
-use criterion::{criterion_group, BatchSize, Criterion, SamplingMode, Throughput};
-use futures::{stream, SinkExt, StreamExt};
+use criterion::{BatchSize, Criterion, SamplingMode, Throughput, criterion_group};
+use futures::{SinkExt, StreamExt, stream};
 use tempfile::tempdir;
 use tokio::fs::OpenOptions;
 use tokio_util::codec::{BytesCodec, FramedWrite};
 use vector::{
     config, sinks, sources,
     test_util::{random_lines, runtime, start_topology},
 };
-use vector_lib::codecs::{encoding::FramingConfig, TextSerializerConfig};
+use vector_lib::codecs::{TextSerializerConfig, encoding::FramingConfig};
 
 fn benchmark_files_no_partitions(c: &mut Criterion) {
     let num_lines: usize = 10_000;

benches/http.rs

@@ -1,24 +1,23 @@
 use std::net::SocketAddr;
 
-use criterion::{criterion_group, BatchSize, BenchmarkId, Criterion, SamplingMode, Throughput};
+use criterion::{BatchSize, BenchmarkId, Criterion, SamplingMode, Throughput, criterion_group};
 use futures::TryFutureExt;
 use hyper::{
-    service::{make_service_fn, service_fn},
     Body, Response, Server,
+    service::{make_service_fn, service_fn},
 };
 use tokio::runtime::Runtime;
 use vector::{
-    config,
+    Error, config,
     sinks::{
         self,
         util::{BatchConfig, Compression},
     },
     sources,
     template::Template,
     test_util::{next_addr, random_lines, runtime, send_lines, start_topology, wait_for_tcp},
-    Error,
 };
-use vector_lib::codecs::{encoding::FramingConfig, TextSerializerConfig};
+use vector_lib::codecs::{TextSerializerConfig, encoding::FramingConfig};
 
 fn benchmark_http(c: &mut Criterion) {
     let num_lines: usize = 1_000;

benches/languages.rs

@@ -1,8 +1,8 @@
-use criterion::{criterion_group, criterion_main, BatchSize, Criterion, SamplingMode, Throughput};
+use criterion::{BatchSize, Criterion, SamplingMode, Throughput, criterion_group, criterion_main};
 use indoc::indoc;
 use vector::{
     config,
-    test_util::{next_addr, runtime, send_lines, start_topology, wait_for_tcp, CountReceiver},
+    test_util::{CountReceiver, next_addr, runtime, send_lines, start_topology, wait_for_tcp},
 };
 
 criterion_group!(

benches/loki.rs

@@ -1,6 +1,6 @@
 use std::convert::TryFrom;
 
-use criterion::{criterion_group, criterion_main, BatchSize, Criterion};
+use criterion::{BatchSize, Criterion, criterion_group, criterion_main};
 use vector::{sinks::loki::valid_label_name, template::Template};
 
 const VALID: [&str; 4] = ["name", " name ", "bee_bop", "a09b"];

benches/lua.rs

@@ -1,7 +1,7 @@
 use std::pin::Pin;
 
-use criterion::{criterion_group, BatchSize, Criterion, Throughput};
-use futures::{stream, SinkExt, Stream, StreamExt};
+use criterion::{BatchSize, Criterion, Throughput, criterion_group};
+use futures::{SinkExt, Stream, StreamExt, stream};
 use indoc::indoc;
 use transforms::lua::v2::LuaConfig;
 use vector::{

benches/metrics_snapshot.rs

@@ -1,4 +1,4 @@
-use criterion::{criterion_group, BenchmarkId, Criterion};
+use criterion::{BenchmarkId, Criterion, criterion_group};
 
 fn benchmark(c: &mut Criterion) {
     let mut group = c.benchmark_group("metrics_snapshot");

benches/remap.rs

@@ -1,13 +1,13 @@
 use std::collections::HashMap;
 
 use chrono::{DateTime, Utc};
-use criterion::{criterion_group, criterion_main, BatchSize, Criterion};
+use criterion::{BatchSize, Criterion, criterion_group, criterion_main};
 use vector::{
     config::{DataType, TransformOutput},
     event::{Event, LogEvent, Value},
     transforms::{
-        remap::{Remap, RemapConfig},
         SyncTransform, TransformOutputsBuf,
+        remap::{Remap, RemapConfig},
     },
 };
 use vrl::event_path;

benches/template.rs

@@ -1,7 +1,7 @@
 use std::convert::TryFrom;
 
 use chrono::Utc;
-use criterion::{criterion_group, BatchSize, Criterion};
+use criterion::{BatchSize, Criterion, criterion_group};
 use vector::{config::log_schema, event::Event, event::LogEvent};
 
 fn bench_elasticsearch_index(c: &mut Criterion) {

benches/transform/common.rs

@@ -6,7 +6,7 @@ use std::{
     task::{Context, Poll},
 };
 
-use futures::{task::noop_waker, Stream};
+use futures::{Stream, task::noop_waker};
 use vector::event::{Event, LogEvent};
 
 // == Streams ==

benches/transform/dedupe.rs

@@ -1,10 +1,10 @@
 use core::fmt;
 use std::{num::NonZeroUsize, time::Duration};
 
-use crate::common::{consume, FixedLogStream};
+use crate::common::{FixedLogStream, consume};
 use criterion::{
-    criterion_group, measurement::WallTime, BatchSize, BenchmarkGroup, BenchmarkId, Criterion,
-    SamplingMode, Throughput,
+    BatchSize, BenchmarkGroup, BenchmarkId, Criterion, SamplingMode, Throughput, criterion_group,
+    measurement::WallTime,
 };
 use vector::transforms::dedupe::common::{CacheConfig, FieldMatchConfig};
 use vector::transforms::dedupe::config::DedupeConfig;

benches/transform/filter.rs

@@ -1,12 +1,12 @@
 use std::time::Duration;
 
 use criterion::{
-    criterion_group, measurement::WallTime, BatchSize, BenchmarkGroup, Criterion, SamplingMode,
-    Throughput,
+    BatchSize, BenchmarkGroup, Criterion, SamplingMode, Throughput, criterion_group,
+    measurement::WallTime,
 };
 use vector::{
     conditions::Condition,
-    transforms::{filter::Filter, FunctionTransform, OutputBuffer},
+    transforms::{FunctionTransform, OutputBuffer, filter::Filter},
 };
 use vector_lib::event::{Event, LogEvent};
 

benches/transform/reduce.rs

@@ -1,10 +1,10 @@
 use core::fmt;
 use std::{num::NonZeroUsize, time::Duration};
 
-use crate::common::{consume, FixedLogStream};
+use crate::common::{FixedLogStream, consume};
 use criterion::{
-    criterion_group, measurement::WallTime, BatchSize, BenchmarkGroup, BenchmarkId, Criterion,
-    SamplingMode, Throughput,
+    BatchSize, BenchmarkGroup, BenchmarkId, Criterion, SamplingMode, Throughput, criterion_group,
+    measurement::WallTime,
 };
 use indexmap::IndexMap;
 use vector::transforms::reduce::config::ReduceConfig;

benches/transform/route.rs

@@ -3,13 +3,13 @@ use std::time::Duration;
 
 use bytes::Bytes;
 use criterion::{
-    criterion_group, measurement::WallTime, BatchSize, BenchmarkGroup, BenchmarkId, Criterion,
-    SamplingMode, Throughput,
+    BatchSize, BenchmarkGroup, BenchmarkId, Criterion, SamplingMode, Throughput, criterion_group,
+    measurement::WallTime,
 };
 use vector::config::TransformContext;
 use vector::transforms::{
-    route::{Route, RouteConfig},
     TransformOutputsBuf,
+    route::{Route, RouteConfig},
 };
 use vector_lib::{
     config::{DataType, TransformOutput},

src/api/handler.rs

@@ -1,10 +1,10 @@
 use std::sync::{
-    atomic::{self, AtomicBool},
     Arc,
+    atomic::{self, AtomicBool},
 };
 
 use serde_json::json;
-use warp::{reply::json, Rejection, Reply};
+use warp::{Rejection, Reply, reply::json};
 
 // Health handler, responds with '{ ok: true }' when running and '{ ok: false}'
 // when shutting down

src/api/schema/components/mod.rs

@@ -10,7 +10,7 @@ use std::{
 };
 
 use async_graphql::{Enum, InputObject, Interface, Object, Subscription};
-use tokio_stream::{wrappers::BroadcastStream, Stream, StreamExt};
+use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream};
 use vector_lib::internal_event::DEFAULT_OUTPUT;
 
 use crate::{
@@ -19,7 +19,7 @@ use crate::{
         filter::{self, filter_items},
         relay, sort,
     },
-    config::{get_transform_output_ids, ComponentKey, Config},
+    config::{ComponentKey, Config, get_transform_output_ids},
     filter_check,
 };
 

src/api/schema/components/sink.rs

@@ -2,7 +2,7 @@ use std::cmp;
 
 use async_graphql::{Enum, InputObject, Object};
 
-use super::{source, state, transform, Component};
+use super::{Component, source, state, transform};
 use crate::{
     api::schema::{
         filter,

src/api/schema/components/source.rs

@@ -2,11 +2,11 @@ use std::cmp;
 
 use async_graphql::{Enum, InputObject, Object};
 
-use super::{sink, state, transform, Component};
+use super::{Component, sink, state, transform};
 use crate::{
     api::schema::{
         filter,
-        metrics::{self, outputs_by_component_key, IntoSourceMetrics, Output},
+        metrics::{self, IntoSourceMetrics, Output, outputs_by_component_key},
         sort,
     },
     config::{ComponentKey, DataType, OutputId},

src/api/schema/components/state.rs

@@ -3,7 +3,7 @@ use std::{
     sync::{Arc, LazyLock, RwLock},
 };
 
-use super::{sink, source, transform, Component};
+use super::{Component, sink, source, transform};
 use crate::config::{ComponentKey, OutputId};
 
 pub const INVARIANT: &str = "Couldn't acquire lock on Vector components. Please report this.";

src/api/schema/components/transform.rs

@@ -2,11 +2,11 @@ use std::cmp;
 
 use async_graphql::{Enum, InputObject, Object};
 
-use super::{sink, source, state, Component};
+use super::{Component, sink, source, state};
 use crate::{
     api::schema::{
         filter,
-        metrics::{self, outputs_by_component_key, IntoTransformMetrics, Output},
+        metrics::{self, IntoTransformMetrics, Output, outputs_by_component_key},
         sort,
     },
     config::{ComponentKey, Inputs, OutputId},

src/api/schema/events/metric.rs

@@ -126,10 +126,14 @@ impl Metric {
                     .expect("logfmt serialization of metric event failed: conversion to serde Value failed. Please report.");
                 match json {
                     Value::Object(map) => encode_logfmt::encode_map(
-                        &map.into_iter().map(|(k,v)| (event::KeyString::from(k), v)).collect(),
+                        &map.into_iter()
+                            .map(|(k, v)| (event::KeyString::from(k), v))
+                            .collect(),
                     )
                     .expect("logfmt serialization of metric event failed. Please report."),
-                    _ => panic!("logfmt serialization of metric event failed: metric converted to unexpected serde Value. Please report."),
+                    _ => panic!(
+                        "logfmt serialization of metric event failed: metric converted to unexpected serde Value. Please report."
+                    ),
                 }
             }
         }

src/api/schema/events/mod.rs

@@ -6,9 +6,9 @@ pub mod trace;
 
 use async_graphql::{Context, Subscription};
 use encoding::EventEncodingType;
-use futures::{stream, Stream, StreamExt};
-use output::{from_tap_payload_to_output_events, OutputEventsPayload};
-use rand::{rngs::SmallRng, Rng, SeedableRng};
+use futures::{Stream, StreamExt, stream};
+use output::{OutputEventsPayload, from_tap_payload_to_output_events};
+use rand::{Rng, SeedableRng, rngs::SmallRng};
 use std::time::{SystemTime, UNIX_EPOCH};
 use tokio::{select, sync::mpsc, time};
 use tokio_stream::wrappers::ReceiverStream;