[pull] master from vectordotdev:master

Cargo.toml

@@ -30,6 +30,7 @@ debug = true
 
 [package.metadata.deb]
 name = "vector"
+section = "admin"
 maintainer-scripts = "distribution/debian/scripts/"
 conf-files = ["/etc/vector/vector.toml", "/etc/default/vector"]
 assets = [
@@ -114,7 +115,7 @@ futures = { version = "0.3.17", default-features = false, features = ["compat",
 tokio = { version = "1.11.0", default-features = false, features = ["full"] }
 tokio-openssl = { version = "0.6.2", default-features = false }
 tokio-stream = { version = "0.1.7", default-features = false, features = ["net", "sync"] }
-tokio-util = { version = "0.6.8", default-features = false, features = ["codec", "time"] }
+tokio-util = { version = "0.6.8", default-features = false, features = ["codec", "time", "io"] }
 
 # Tracing
 tracing = { version = "0.1.27", default-features = false }
@@ -212,7 +213,8 @@ dyn-clone = { version = "1.0.4", default-features = false }
 encoding_rs = { version = "0.8.28", features = ["serde"] }
 evmap = { version = "10.0.2", default-features = false, optional = true }
 exitcode = { version = "1.1.2", default-features = false }
-flate2 = { version = "1.0.22", default-features = false }
+flate2 = { version = "1.0.21", default-features = false }
+futures-util = { version = "0.3.17", default-features = false }
 getset = { version = "0.1.1", default-features = false }
 glob = { version = "0.3.0", default-features = false }
 grok = { version = "1.2.0", default-features = false, optional = true }
@@ -263,7 +265,7 @@ semver = { version = "1.0.4", default-features = false, features = ["serde", "st
 smallvec = { version = "1", optional = true }
 snafu = { version = "0.6.10", default-features = false, features = ["futures"] }
 snap = { version = "1.0.5", default-features = false, optional = true }
-socket2 = { version = "0.4.1", default-features = false }
+socket2 = { version = "0.4.2", default-features = false }
 stream-cancel = { version = "0.8.1", default-features = false }
 strip-ansi-escapes = { version = "0.1.1", default-features = false }
 structopt = { version = "0.3.23", default-features = false }
@@ -302,7 +304,7 @@ approx = "0.5.0"
 assert_cmd = "2.0.1"
 base64 = "0.13.0"
 criterion = { version = "0.3.5", features = ["html_reports", "async_tokio"] }
-libc = "0.2.101"
+libc = "0.2.102"
 libz-sys = "1.1.3"
 matches = "0.1.9"
 pretty_assertions = "0.7.2"
@@ -467,7 +469,7 @@ sources-prometheus = ["prometheus-parser", "sinks-prometheus", "sources-utils-ht
 sources-socket = ["bytesize", "listenfd", "tokio-util/net", "sources-utils-udp", "sources-utils-tcp-keepalive", "sources-utils-tcp-socket", "sources-utils-tls", "sources-utils-unix", "codecs"]
 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/net", "codecs"]
-sources-stdin = ["bytesize"]
+sources-stdin = ["bytesize", "codecs"]
 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", "codecs"]
 sources-utils-http = ["snap", "sources-utils-tls", "warp"]
 sources-utils-tcp-keepalive = []
@@ -569,6 +571,7 @@ sinks-logs = [
   "sinks-clickhouse",
   "sinks-console",
   "sinks-datadog",
+  "sinks-datadog_archives",
   "sinks-elasticsearch",
   "sinks-file",
   "sinks-gcp",
@@ -616,6 +619,7 @@ sinks-blackhole = []
 sinks-clickhouse = ["bytesize"]
 sinks-console = []
 sinks-datadog = ["bytesize"]
+sinks-datadog_archives = ["sinks-aws_s3"]
 sinks-elasticsearch = ["bytesize", "rusoto", "transforms-metric_to_log"]
 sinks-file = []
 sinks-gcp = ["base64", "bytesize", "goauth", "gouth", "smpl_jwt", "uuid"]

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

@@ -260,7 +260,8 @@ pub enum BatchStatus {
 }
 
 impl BatchStatus {
-    /// Update this status with another batch's delivery status, and return the result.
+    /// Update this status with another batch's delivery status, and return the
+    /// result.
     #[allow(clippy::match_same_arms)] // False positive: https://github.com/rust-lang/rust-clippy/issues/860
     fn update(self, status: EventStatus) -> Self {
         match (self, status) {
@@ -281,21 +282,24 @@ impl BatchStatus {
 #[derivative(Default)]
 #[repr(u8)]
 pub enum EventStatus {
-    /// All copies of this event were dropped without being finalized (the default).
+    /// All copies of this event were dropped without being finalized (the
+    /// default).
     #[derivative(Default)]
     Dropped,
     /// All copies of this event were delivered successfully.
     Delivered,
     /// At least one copy of this event encountered a retriable error.
     Errored,
-    /// At least one copy of this event encountered a permanent failure or rejection.
+    /// At least one copy of this event encountered a permanent failure or
+    /// rejection.
     Failed,
     /// This status has been recorded and should not be updated.
     Recorded,
 }
 
 impl EventStatus {
-    /// Update this status with another event's finalization status and return the result.
+    /// Update this status with another event's finalization status and return
+    /// the result.
     ///
     /// # Panics
     ///

src/http.rs

@@ -149,7 +149,7 @@ impl<B> Service<Request<B>> for HttpClient<B>
 where
     B: fmt::Debug + HttpBody + Send + 'static,
     B::Data: Send,
-    B::Error: Into<crate::Error>,
+    B::Error: Into<crate::Error> + Send,
 {
     type Response = http::Response<Body>;
     type Error = HttpError;

src/internal_events/common.rs

@@ -32,6 +32,7 @@ impl InternalEvent for EventsSent {
 
     fn emit_metrics(&self) {
         if self.count > 0 {
+            counter!("events_out_total", self.count as u64);
             counter!("sent_events_total", self.count as u64);
             counter!("sent_event_bytes_total", self.byte_size as u64);
         }

src/internal_events/stdin.rs

@@ -2,33 +2,19 @@ use super::InternalEvent;
 use metrics::counter;
 
 #[derive(Debug)]
-pub struct StdinEventReceived {
+pub struct StdinEventsReceived {
     pub byte_size: usize,
+    pub count: usize,
 }
 
-impl InternalEvent for StdinEventReceived {
+impl InternalEvent for StdinEventsReceived {
     fn emit_logs(&self) {
-        trace!(message = "Received one event.");
+        trace!(message = "Received events.", self.count);
     }
 
     fn emit_metrics(&self) {
-        counter!("received_events_total", 1);
-        counter!("events_in_total", 1);
+        counter!("received_events_total", self.count as u64);
+        counter!("events_in_total", self.count as u64);
         counter!("processed_bytes_total", self.byte_size as u64);
     }
 }
-
-#[derive(Debug)]
-pub struct StdinReadFailed {
-    pub error: std::io::Error,
-}
-
-impl InternalEvent for StdinReadFailed {
-    fn emit_logs(&self) {
-        error!(message = "Unable to read from source.", error = ?self.error);
-    }
-
-    fn emit_metrics(&self) {
-        counter!("stdin_reads_failed_total", 1);
-    }
-}