vectordotdev · pablosichert · Apr 2, 2021 · Jan 5, 2021 · Jan 11, 2021 · Jan 11, 2021
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -77,12 +77,13 @@ vector-api-client = { path = "lib/vector-api-client", optional = true }
 vrl-cli = { path = "lib/vrl/cli", optional = true }
 
 # Tokio / Futures
-async-trait = "0.1"
+async-trait = "0.1.42"
 futures = { version = "0.3", default-features = false, features = ["compat", "io-compat"] }
 futures01 = { package = "futures", version = "0.1.25" }
-tokio = { version = "0.2.13", features = ["blocking", "fs", "io-std", "macros", "process", "rt-core", "rt-threaded", "signal", "stream", "sync", "time", "udp", "uds"] }
-tokio-openssl = "0.4.0"
-tokio-util = { version = "0.3.1", features = ["codec"] }
+tokio = { version = "1.3.0", features = ["full"] }
+tokio-openssl = "0.6.1"
+tokio-stream = { version = "0.1.2", features = ["net"] }
+tokio-util = { version = "0.6.2", features = ["codec", "time"] }
 
 # Tracing
 tracing = "0.1.15"
@@ -101,37 +102,38 @@ metrics-util = "=0.4.0-alpha.10"
 metrics-macros = "=0.1.0-alpha.9"
 
 # Aws
-rusoto_cloudwatch = { version = "0.45.0", optional = true }
-rusoto_core = { version = "0.45.0", features = ["encoding"], optional = true }
-rusoto_credential = { version = "0.45.0", optional = true }
-rusoto_es = { version = "0.45.0", optional = true }
-rusoto_firehose = { version = "0.45.0", optional = true }
-rusoto_kinesis = { version = "0.45.0", optional = true }
-rusoto_logs = { version = "0.45.0", optional = true }
-rusoto_s3 = { version = "0.45.0", optional = true }
-rusoto_signature = { version = "0.45.0", optional = true }
-rusoto_sqs = { version = "0.45.0", optional = true }
-rusoto_sts = { version = "0.45.0", optional = true }
+rusoto_cloudwatch = { version = "0.46.0", optional = true }
+rusoto_core = { version = "0.46.0", features = ["encoding"], optional = true }
+rusoto_credential = { version = "0.46.0", optional = true }
+rusoto_es = { version = "0.46.0", optional = true }
+rusoto_firehose = { version = "0.46.0", optional = true }
+rusoto_kinesis = { version = "0.46.0", optional = true }
+rusoto_logs = { version = "0.46.0", optional = true }
+rusoto_s3 = { version = "0.46.0", optional = true }
+rusoto_signature = { version = "0.46.0", optional = true }
+rusoto_sqs = { version = "0.46.0", optional = true }
+rusoto_sts = { version = "0.46.0", optional = true }
 
 # Tower
-tower = { version = "0.3.1", git = "https://github.com/tower-rs/tower", rev = "43168944220ed32dab83cb4f11f7b97abc5818d5", features = ["buffer", "limit", "retry", "timeout", "util"] }
+tower = { version = "0.4.0", features = ["buffer", "limit", "retry", "timeout", "util"] }
+tower-layer = "0.3.1"
 
 # Serde
 serde = { version = "1.0.117", features = ["derive"] }
 serde_json = { version = "1.0.33", features = ["raw_value"] }
 serde_yaml = { version ="0.8.13" }
 
 # Prost
-prost = "0.6.1"
-prost-types = "0.6.1"
+prost = "0.7.0"
+prost-types = "0.7.0"
 
 # GCP
 goauth = { version = "0.9.0", optional = true }
 smpl_jwt = { version = "0.6.1", optional = true }
 
 # API
-async-graphql = { version = "=2.5.0", optional = true }
-async-graphql-warp = { version = "=2.5.0", optional = true }
+async-graphql = { version = "2.5.9", optional = true }
+async-graphql-warp = { version = "2.5.9", optional = true }
 itertools = { version = "0.10.0", optional = true }
 
 # API client
@@ -146,12 +148,12 @@ vrl-stdlib = { path = "lib/vrl/stdlib" }
 
 # External libs
 anyhow = "1.0.37"
-async-compression = { version = "0.3.7", features = ["tokio-02", "gzip", "zstd"] }
+async-compression = { version = "0.3.7", features = ["tokio", "gzip", "zstd"] }
 avro-rs = { version = "0.13.0", optional = true }
 base64 = { version = "0.13.0", optional = true }
 bloom = { version = "0.3.2", optional = true }
-bollard = { version = "0.9.1", features = ["ssl"], optional = true }
-bytes = { version = "0.5.6", features = ["serde"] }
+bollard = { version = "0.10.1", features = ["ssl"], optional = true }
+bytes = { version = "1.0.0", features = ["serde"] }
 bytesize = { version = "1.0.0", optional = true }
 chrono = { version = "0.4.19", features = ["serde"] }
 cidr-utils = "0.5.0"
@@ -162,7 +164,7 @@ derivative = "2.1.1"
 dirs-next = { version = "2.0.0", optional = true }
 dyn-clone = "1.0.3"
 encoding_rs = { version = "0.8", features = ["serde"] }
-evmap = { version = "10.0.2", features = ["bytes"], optional = true }
+evmap = { git = "https://github.com/lukesteensen/evmap.git", rev = "45ba973c22715a68c5e99efad4b072421f7ad40b", features = ["bytes"], optional = true }
 exitcode = "1.1.2"
 flate2 = "1.0.19"
 getset = "0.1.1"
@@ -172,8 +174,8 @@ headers = "0.3"
 heim = { version = "0.1.0-rc.1", features = ["full"], optional = true }
 hostname = "0.3.1"
 http = "0.2"
-hyper = "0.13"
-hyper-openssl = "0.8"
+hyper = { version = "0.14", features = ["full"] }
+hyper-openssl = "0.9.1"
 indexmap = {version = "1.6.2", features = ["serde"]}
 indoc = "1.0.3"
 inventory = "0.1.10"
@@ -201,16 +203,17 @@ postgres-openssl = { version = "0.3.0", optional = true }
 pulsar = { version = "1.0.0", default-features = false, features = ["tokio-runtime"], optional = true }
 rand = { version = "0.8.0", features = ["small_rng"] }
 rand_distr = "0.4.0"
-rdkafka = { version = "0.24.0", features = ["libz", "ssl", "zstd"], optional = true }
+# Move to 0.25.x release after fix for aarch64 builds has been published: https://github.com/fede1024/rust-rdkafka/pull/346.
+rdkafka = { git = "https://github.com/fede1024/rust-rdkafka", rev = "52bcef43b684f90294d8b4b92a5e6b1129aab468", features = ["libz", "ssl", "zstd"], optional = true }
 regex = "1.3.9"
 # make sure to update the external docs when the Lua version changes
 rlua = { version = "0.17.0", optional = true }
 seahash = { version = "4.0.1", optional = true }
 semver = { version = "0.11.0", features = ["serde"], optional = true }
 snafu = { version = "0.6.10", features = ["futures", "futures-01"] }
 snap = { version = "1.0.3", optional = true }
-socket2 = { version = "0.3.19", optional = true }
-stream-cancel = "0.6.2"
+socket2 = "0.4.0"
+stream-cancel = "0.8.0"
 strip-ansi-escapes = "0.1.0"
 structopt = "0.3.21"
 syslog = { version = "5", optional = true }
@@ -223,7 +226,7 @@ toml = "0.5.8"
 typetag = "0.1.6"
 url = "2.2.1"
 uuid = { version = "0.8", features = ["serde", "v4"], optional = true }
-warp = { version = "0.2.5", default-features = false, optional = true }
+warp = { version = "0.3.0", default-features = false, optional = true }
 
 # For WASM
 async-stream = "0.3.0"
@@ -244,7 +247,7 @@ atty = "0.2"
 nix = "0.19.0"
 
 [build-dependencies]
-prost-build = "0.6.1"
+prost-build = "0.7.0"
 built = { version = "0.4.4", features = ["chrono", "git2"] }
 
 [dev-dependencies]
@@ -257,13 +260,13 @@ libc = "0.2.80"
 libz-sys = "1.1.2"
 matches = "0.1.8"
 pretty_assertions = "0.7.1"
-reqwest = { version = "0.10.9", features = ["json"] }
+reqwest = { version = "0.11.0", features = ["json"] }
 rusty-fork = "0.3.0"
 tempfile = "3.0.6"
-tokio = { version = "0.2", features = ["test-util"] }
-tokio-test = "0.4"
+tokio = { version = "1.3.0", features = ["test-util"] }
+tokio-test = "0.4.0"
 tokio01-test = "0.1.1"
-tower-test = "0.3.0"
+tower-test = "0.4.0"
 walkdir = "2.2.7"
 
 [features]
@@ -383,16 +386,16 @@ sources-mongodb_metrics = ["mongodb"]
 sources-nginx_metrics = ["nom"]
 sources-postgresql_metrics = ["postgres-openssl", "tokio-postgres"]
 sources-prometheus = ["prometheus-parser", "sinks-prometheus", "sources-utils-http", "warp"]
-sources-socket = ["bytesize", "listenfd", "tokio-util/udp", "sources-utils-udp", "sources-utils-tcp-keepalive", "sources-utils-tcp-socket", "sources-utils-tls", "sources-utils-unix"]
+sources-socket = ["bytesize", "listenfd", "tokio-util/net", "sources-utils-udp", "sources-utils-tcp-keepalive", "sources-utils-tcp-socket", "sources-utils-tls", "sources-utils-unix"]
 sources-splunk_hec = ["bytesize", "sources-utils-tls", "warp"]
-sources-statsd = ["listenfd", "sources-utils-tcp-keepalive", "sources-utils-tcp-socket", "sources-utils-tls", "sources-utils-udp", "sources-utils-unix", "tokio-util/udp"]
+sources-statsd = ["listenfd", "sources-utils-tcp-keepalive", "sources-utils-tcp-socket", "sources-utils-tls", "sources-utils-udp", "sources-utils-unix", "tokio-util/net"]
 sources-stdin = ["bytesize"]
-sources-syslog = ["bytesize", "listenfd", "tokio-util/udp", "sources-utils-udp", "sources-utils-tcp-keepalive", "sources-utils-tcp-socket", "sources-utils-tls", "sources-utils-unix", "syslog_loose"]
+sources-syslog = ["bytesize", "listenfd", "tokio-util/net", "sources-utils-udp", "sources-utils-tcp-keepalive", "sources-utils-tcp-socket", "sources-utils-tls", "sources-utils-unix", "syslog_loose"]
 sources-utils-http = ["snap", "sources-utils-tls", "warp"]
 sources-utils-tcp-keepalive = []
 sources-utils-tcp-socket = []
 sources-utils-tls = []
-sources-utils-udp = ["socket2"]
+sources-utils-udp = []
 sources-utils-unix = []
 sources-vector = ["listenfd", "sources-utils-tcp-keepalive", "sources-utils-tcp-socket", "sources-utils-tls"]
 
@@ -545,8 +548,8 @@ sinks-pulsar = ["avro-rs", "pulsar"]
 sinks-sematext = ["sinks-elasticsearch", "sinks-influxdb"]
 sinks-socket = ["sinks-utils-udp"]
 sinks-splunk_hec = ["bytesize"]
-sinks-statsd = ["sinks-utils-udp", "tokio-util/udp"]
-sinks-utils-udp = ["socket2"]
+sinks-statsd = ["sinks-utils-udp", "tokio-util/net"]
+sinks-utils-udp = []
 sinks-vector = ["sinks-utils-udp"]
 
 # Identifies that the build is a nightly build
@@ -664,10 +667,3 @@ required-features = ["metrics-benches"]
 name = "distribution_statistic"
 harness = false
 required-features = ["statistic-benches"]
-
-[patch.'https://github.com/tower-rs/tower']
-tower-layer = "=0.3.0"
-
-[patch.crates-io]
-# TODO: update to the next 0.13.x (after 0.13.10, if any) or 0.14 (or higher)
-hyper = { version = "0.13", git = "https://github.com/hyperium/hyper", rev = "d7495a75abca34646b1d6d047589c1b8110d0fa5" }
diff --git a/benches/batch.rs b/benches/batch.rs
@@ -61,7 +61,7 @@ fn benchmark_batching(c: &mut Criterion) {
                             batch_sink,
                         )
                     },
-                    |(mut rt, input, batch_sink)| rt.block_on(input.forward(batch_sink)).unwrap(),
+                    |(rt, input, batch_sink)| rt.block_on(input.forward(batch_sink)).unwrap(),
                     criterion::BatchSize::LargeInput,
                 )
             },
@@ -86,7 +86,7 @@ fn benchmark_batching(c: &mut Criterion) {
 
                     (rt, stream::iter(input.clone()).map(Ok), batch_sink)
                 },
-                |(mut rt, input, batch_sink)| rt.block_on(input.forward(batch_sink)).unwrap(),
+                |(rt, input, batch_sink)| rt.block_on(input.forward(batch_sink)).unwrap(),
                 criterion::BatchSize::LargeInput,
             )
         });

diff --git a/benches/buffering.rs b/benches/buffering.rs
@@ -34,7 +34,7 @@ fn benchmark_buffers(c: &mut Criterion) {
                     when_full: Default::default(),
                 };
 
-                let mut rt = runtime();
+                let rt = runtime();
                 let (output_lines, topology) = rt.block_on(async move {
                     let output_lines = CountReceiver::receive_lines(out_addr);
                     let (topology, _crash) = start_topology(config.build().unwrap(), false).await;
@@ -44,7 +44,7 @@ fn benchmark_buffers(c: &mut Criterion) {
 
                 (rt, topology, output_lines)
             },
-            |(mut rt, topology, output_lines)| {
+            |(rt, topology, output_lines)| {
                 rt.block_on(async move {
                     let lines = random_lines(line_size).take(num_lines);
                     send_lines(in_addr, lines).await.unwrap();
@@ -82,7 +82,7 @@ fn benchmark_buffers(c: &mut Criterion) {
                     when_full: Default::default(),
                 };
                 config.global.data_dir = Some(data_dir.path().to_path_buf());
-                let mut rt = runtime();
+                let rt = runtime();
                 let (output_lines, topology) = rt.block_on(async move {
                     let output_lines = CountReceiver::receive_lines(out_addr);
                     let (topology, _crash) = start_topology(config.build().unwrap(), false).await;
@@ -91,7 +91,7 @@ fn benchmark_buffers(c: &mut Criterion) {
                 });
                 (rt, topology, output_lines)
             },
-            |(mut rt, topology, output_lines)| {
+            |(rt, topology, output_lines)| {
                 rt.block_on(async move {
                     let lines = random_lines(line_size).take(num_lines);
                     send_lines(in_addr, lines).await.unwrap();
@@ -130,7 +130,7 @@ fn benchmark_buffers(c: &mut Criterion) {
     //when_full: Default::default(),
     //};
     //config.global.data_dir = Some(data_dir.path().to_path_buf());
-    //let mut rt = runtime();
+    //let rt = runtime();
     //let (output_lines, topology) = rt.block_on(async move {
     //let output_lines = CountReceiver::receive_lines(out_addr);
     //let (topology, _crash) = start_topology(config.build().unwrap(), false).await;
@@ -139,7 +139,7 @@ fn benchmark_buffers(c: &mut Criterion) {
     //});
     //(rt, topology, output_lines)
     //},
-    //|(mut rt, topology, output_lines)| {
+    //|(rt, topology, output_lines)| {
     //rt.block_on(async move {
     //let lines = random_lines(line_size).take(num_lines);
     //send_lines(in_addr, lines).await.unwrap();

diff --git a/benches/files.rs b/benches/files.rs
@@ -59,7 +59,7 @@ fn benchmark_files_without_partitions(c: &mut Criterion) {
                     },
                 );
 
-                let mut rt = runtime();
+                let rt = runtime();
                 let (topology, input) = rt.block_on(async move {
                     let (topology, _crash) = start_topology(config.build().unwrap(), false).await;
 
@@ -74,7 +74,7 @@ fn benchmark_files_without_partitions(c: &mut Criterion) {
                 });
                 (rt, topology, input)
             },
-            |(mut rt, topology, input)| {
+            |(rt, topology, input)| {
                 rt.block_on(async move {
                     let lines = random_lines(line_size).take(num_lines).map(|mut line| {
                         line.push('\n');

diff --git a/benches/http.rs b/benches/http.rs
@@ -58,7 +58,7 @@ fn benchmark_http(c: &mut Criterion) {
                             },
                         );
 
-                        let mut rt = runtime();
+                        let rt = runtime();
                         let topology = rt.block_on(async move {
                             let (topology, _crash) =
                                 start_topology(config.build().unwrap(), false).await;
@@ -67,7 +67,7 @@ fn benchmark_http(c: &mut Criterion) {
                         });
                         (rt, topology)
                     },
-                    |(mut rt, topology)| {
+                    |(rt, topology)| {
                         rt.block_on(async move {
                             let lines = random_lines(line_size).take(num_lines);
                             send_lines(in_addr, lines).await.unwrap();

diff --git a/benches/isolated_buffering.rs b/benches/isolated_buffering.rs
@@ -7,6 +7,7 @@ use futures::{
 };
 use futures01::{stream, Sink, Stream};
 use tempfile::tempdir;
+use tokio_stream::wrappers::ReceiverStream;
 use vector::{
     buffers::{
         disk::{leveldb_buffer, DiskBuffer},
@@ -55,7 +56,7 @@ fn benchmark_buffers(c: &mut Criterion) {
 
                 (rt, writer, read_loop)
             },
-            |(mut rt, writer, read_loop)| {
+            |(rt, writer, read_loop)| {
                 let send = writer.send_all(random_events(line_size).take(num_lines as u64));
 
                 let read_handle = rt.spawn(read_loop.compat());
@@ -81,7 +82,7 @@ fn benchmark_buffers(c: &mut Criterion) {
 
                 (rt, writer, read_handle)
             },
-            |(mut rt, mut writer, read_handle)| {
+            |(rt, mut writer, read_handle)| {
                 let write_handle = rt.spawn(async move {
                     let mut stream = random_events(line_size).take(num_lines as u64).compat();
                     while let Some(e) = stream.next().await {
@@ -101,13 +102,14 @@ fn benchmark_buffers(c: &mut Criterion) {
             || {
                 let rt = runtime();
 
-                let (writer, mut reader) = tokio::sync::mpsc::channel(100);
+                let (writer, reader) = tokio::sync::mpsc::channel(100);
+                let mut stream = ReceiverStream::new(reader);
 
-                let read_handle = rt.spawn(async move { while reader.next().await.is_some() {} });
+                let read_handle = rt.spawn(async move { while stream.next().await.is_some() {} });
 
                 (rt, writer, read_handle)
             },
-            |(mut rt, mut writer, read_handle)| {
+            |(rt, writer, read_handle)| {
                 let write_handle = rt.spawn(async move {
                     let mut stream = random_events(line_size).take(num_lines as u64).compat();
                     while let Some(e) = stream.next().await {
@@ -136,7 +138,7 @@ fn benchmark_buffers(c: &mut Criterion) {
 
                 (rt, writer)
             },
-            |(mut rt, writer)| {
+            |(rt, writer)| {
                 let send = writer.send_all(random_events(line_size).take(num_lines as u64));
                 let write_handle = rt.spawn(send.compat());
                 let _ = rt.block_on(write_handle).unwrap().unwrap();
@@ -150,7 +152,7 @@ fn benchmark_buffers(c: &mut Criterion) {
             || {
                 let data_dir = tempdir().unwrap();
 
-                let mut rt = runtime();
+                let rt = runtime();
 
                 let plenty_of_room = num_lines * line_size * 2;
                 let (writer, reader, acker) =
@@ -170,7 +172,7 @@ fn benchmark_buffers(c: &mut Criterion) {
 
                 (rt, read_loop)
             },
-            |(mut rt, read_loop)| {
+            |(rt, read_loop)| {
                 let read_handle = rt.spawn(read_loop);
                 rt.block_on(read_handle).unwrap().unwrap();
             },
@@ -198,7 +200,7 @@ fn benchmark_buffers(c: &mut Criterion) {
 
                 (rt, writer, read_loop)
             },
-            |(mut rt, writer, read_loop)| {
+            |(rt, writer, read_loop)| {
                 let send = writer.send_all(random_events(line_size).take(num_lines as u64));
 
                 let read_handle = rt.spawn(read_loop);