fix(observability, sinks): Fix missing increment in batch counter and add component tests to HttpSink sinks

src/sinks/clickhouse.rs

@@ -281,13 +281,13 @@ mod integration_tests {
     use crate::{
         config::{log_schema, SinkConfig, SinkContext},
         sinks::util::encoding::TimestampFormat,
+        test_util::components::{self, HTTP_SINK_TAGS},
         test_util::{random_string, trace_init},
     };
     use futures::{future, stream};
     use serde_json::Value;
     use std::{
         convert::Infallible,
-        future::ready,
         net::SocketAddr,
         sync::{
             atomic::{AtomicBool, Ordering},
@@ -335,9 +335,7 @@ mod integration_tests {
             .as_mut_log()
             .insert("items", vec!["item1", "item2"]);
 
-        sink.run(stream::once(ready(input_event.clone())))
-            .await
-            .unwrap();
+        components::run_sink_event(sink, input_event.clone(), &HTTP_SINK_TAGS).await;
 
         let output = client.select_all(&table).await;
         assert_eq!(1, output.rows);
@@ -381,9 +379,7 @@ mod integration_tests {
         let (mut input_event, mut receiver) = make_event();
         input_event.as_mut_log().insert("unknown", "mysteries");
 
-        sink.run(stream::once(ready(input_event.clone())))
-            .await
-            .unwrap();
+        components::run_sink_event(sink, input_event.clone(), &HTTP_SINK_TAGS).await;
 
         let output = client.select_all(&table).await;
         assert_eq!(1, output.rows);
@@ -434,9 +430,7 @@ mod integration_tests {
 
         let (mut input_event, _receiver) = make_event();
 
-        sink.run(stream::once(future::ready(input_event.clone())))
-            .await
-            .unwrap();
+        components::run_sink_event(sink, input_event.clone(), &HTTP_SINK_TAGS).await;
 
         let output = client.select_all(&table).await;
         assert_eq!(1, output.rows);
@@ -493,9 +487,7 @@ timestamp_format = "unix""#,
 
         let (mut input_event, _receiver) = make_event();
 
-        sink.run(stream::once(future::ready(input_event.clone())))
-            .await
-            .unwrap();
+        components::run_sink_event(sink, input_event.clone(), &HTTP_SINK_TAGS).await;
 
         let output = client.select_all(&table).await;
         assert_eq!(1, output.rows);

src/sinks/datadog/events.rs

@@ -268,6 +268,7 @@ mod tests {
     use crate::{
         config::SinkConfig,
         sinks::util::test::{build_test_server_status, load_sink},
+        test_util::components::{self, HTTP_SINK_TAGS},
         test_util::{next_addr, random_lines_with_stream},
     };
     use bytes::Bytes;
@@ -325,7 +326,11 @@ mod tests {
         let (batch, mut receiver) = BatchNotifier::new_with_receiver();
         let (expected, events) = random_events_with_stream(100, 10, Some(batch));
 
-        let _ = sink.run(events).await.unwrap();
+        components::init();
+        sink.run(events).await.unwrap();
+        if batch_status == BatchStatus::Delivered {
+            components::SINK_TESTS.assert(&HTTP_SINK_TAGS);
+        }
 
         assert_eq!(receiver.try_recv(), Ok(batch_status));
 
@@ -385,14 +390,14 @@ mod tests {
 
         let (expected, events) = random_events_with_stream(100, 10, None);
 
-        let mut events = events.map(|mut e| {
+        let events = events.map(|mut e| {
             e.as_mut_log()
                 .metadata_mut()
                 .set_datadog_api_key(Some(Arc::from("from_metadata")));
             Ok(e)
         });
 
-        let _ = sink.into_sink().send_all(&mut events).await.unwrap();
+        components::sink_send_stream(sink, events, &HTTP_SINK_TAGS).await;
         let output = rx.take(expected.len()).collect::<Vec<_>>().await;
 
         for (i, val) in output.iter().enumerate() {

src/sinks/elasticsearch/mod.rs

@@ -1107,15 +1107,16 @@ mod integration_tests {
         config::{ProxyConfig, SinkConfig, SinkContext},
         http::HttpClient,
         sinks::HealthcheckError,
+        test_util::components::{self, HTTP_SINK_TAGS},
         test_util::{random_events_with_stream, random_string, trace_init},
         tls::{self, TlsOptions},
     };
     use chrono::Utc;
-    use futures::{stream, StreamExt};
+    use futures::StreamExt;
     use http::{Request, StatusCode};
     use hyper::Body;
     use serde_json::{json, Value};
-    use std::{fs::File, future::ready, io::Read};
+    use std::{fs::File, io::Read};
     use vector_core::event::{BatchNotifier, BatchStatus, LogEvent};
 
     impl ElasticSearchCommon {
@@ -1229,9 +1230,7 @@ mod integration_tests {
 
         let timestamp = input_event[crate::config::log_schema().timestamp_key()].clone();
 
-        sink.run(stream::once(ready(input_event.into())))
-            .await
-            .unwrap();
+        components::run_sink_event(sink, input_event.into(), &HTTP_SINK_TAGS).await;
 
         assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));
 
@@ -1446,17 +1445,20 @@ mod integration_tests {
         if break_events {
             // Break all but the first event to simulate some kind of partial failure
             let mut doit = false;
-            sink.run(events.map(move |mut event| {
-                if doit {
-                    event.as_mut_log().insert("_type", 1);
-                }
-                doit = true;
-                event
-            }))
-            .await
-            .expect("Sending events failed");
+            components::run_sink(
+                sink,
+                events.map(move |mut event| {
+                    if doit {
+                        event.as_mut_log().insert("_type", 1);
+                    }
+                    doit = true;
+                    event
+                }),
+                &HTTP_SINK_TAGS,
+            )
+            .await;
         } else {
-            sink.run(events).await.expect("Sending events failed");
+            components::run_sink(sink, events, &HTTP_SINK_TAGS).await;
         }
 
         assert_eq!(receiver.try_recv(), Ok(batch_status));

src/sinks/gcp/pubsub.rs

@@ -221,6 +221,7 @@ mod tests {
 #[cfg(feature = "gcp-pubsub-integration-tests")]
 mod integration_tests {
     use super::*;
+    use crate::test_util::components::{self, HTTP_SINK_TAGS};
     use crate::test_util::{random_events_with_stream, random_string, trace_init};
     use reqwest::{Client, Method, Response};
     use serde_json::{json, Value};
@@ -255,7 +256,7 @@ mod integration_tests {
 
         let (batch, mut receiver) = BatchNotifier::new_with_receiver();
         let (input, events) = random_events_with_stream(100, 100, Some(batch));
-        sink.run(events).await.expect("Sending events failed");
+        components::run_sink(sink, events, &HTTP_SINK_TAGS).await;
         assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));
 
         let response = pull_messages(&subscription, 1000).await;

src/sinks/http.rs

@@ -312,7 +312,7 @@ mod tests {
                 test::{build_test_server, build_test_server_generic, build_test_server_status},
             },
         },
-        test_util::{next_addr, random_lines_with_stream},
+        test_util::{components, components::HTTP_SINK_TAGS, next_addr, random_lines_with_stream},
     };
     use bytes::{Buf, Bytes};
     use flate2::read::MultiGzDecoder;
@@ -664,13 +664,11 @@ mod tests {
         let (in_addr, sink) = build_sink(extra_config).await;
 
         let (rx, trigger, server) = build_test_server(in_addr);
+        tokio::spawn(server);
 
         let (batch, mut receiver) = BatchNotifier::new_with_receiver();
         let (input_lines, events) = random_lines_with_stream(100, num_lines, Some(batch));
-        let pump = sink.run(events);
-
-        tokio::spawn(server);
-        pump.await.unwrap();
+        components::run_sink(sink, events, &HTTP_SINK_TAGS).await;
         drop(trigger);
 
         assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));

src/sinks/humio/logs.rs

@@ -150,13 +150,12 @@ mod integration_tests {
         config::{log_schema, SinkConfig, SinkContext},
         event::Event,
         sinks::util::Compression,
-        test_util::random_string,
+        test_util::{components, components::HTTP_SINK_TAGS, random_string},
     };
     use chrono::Utc;
-    use futures::stream;
     use indoc::indoc;
     use serde_json::{json, Value as JsonValue};
-    use std::{collections::HashMap, convert::TryFrom, future::ready};
+    use std::{collections::HashMap, convert::TryFrom};
 
     // matches humio container address
     const HOST: &str = "http://localhost:8080";
@@ -177,7 +176,7 @@ mod integration_tests {
         let log = event.as_mut_log();
         log.insert(log_schema().host_key(), host.clone());
 
-        sink.run(stream::once(ready(event))).await.unwrap();
+        components::run_sink_event(sink, event, &HTTP_SINK_TAGS).await;
 
         let entry = find_entry(repo.name.as_str(), message.as_str()).await;
 
@@ -213,7 +212,7 @@ mod integration_tests {
 
         let message = random_string(100);
         let event = Event::from(message.clone());
-        sink.run(stream::once(ready(event))).await.unwrap();
+        components::run_sink_event(sink, event, &HTTP_SINK_TAGS).await;
 
         let entry = find_entry(repo.name.as_str(), message.as_str()).await;
 
@@ -246,7 +245,7 @@ mod integration_tests {
                 .as_mut_log()
                 .insert("@timestamp", Utc::now().to_rfc3339());
 
-            sink.run(stream::once(ready(event))).await.unwrap();
+            components::run_sink_event(sink, event, &HTTP_SINK_TAGS).await;
 
             let entry = find_entry(repo.name.as_str(), message.as_str()).await;
 
@@ -269,7 +268,7 @@ mod integration_tests {
             let message = random_string(100);
             let event = Event::from(message.clone());
 
-            sink.run(stream::once(ready(event))).await.unwrap();
+            components::run_sink_event(sink, event, &HTTP_SINK_TAGS).await;
 
             let entry = find_entry(repo.name.as_str(), message.as_str()).await;
 

src/sinks/humio/metrics.rs

@@ -116,7 +116,7 @@ mod tests {
             Event, Metric,
         },
         sinks::util::test::{build_test_server, load_sink},
-        test_util,
+        test_util::{self, components, components::HTTP_SINK_TAGS},
     };
     use chrono::{offset::TimeZone, Utc};
     use indoc::indoc;
@@ -203,7 +203,7 @@ mod tests {
         ];
 
         let len = metrics.len();
-        let _ = sink.run(stream::iter(metrics)).await.unwrap();
+        components::run_sink(sink, stream::iter(metrics), &HTTP_SINK_TAGS).await;
 
         let output = rx.take(len).collect::<Vec<_>>().await;
         assert_eq!(

src/sinks/influxdb/logs.rs

@@ -240,7 +240,7 @@ mod tests {
             http::HttpSink,
             test::{build_test_server_status, load_sink},
         },
-        test_util::next_addr,
+        test_util::{components, components::HTTP_SINK_TAGS, next_addr},
     };
     use chrono::{offset::TimeZone, Utc};
     use futures::{channel::mpsc, stream, StreamExt};
@@ -591,7 +591,11 @@ mod tests {
         }
         drop(batch);
 
+        components::init();
         sink.run(stream::iter(events)).await.unwrap();
+        if batch_status == BatchStatus::Delivered {
+            components::SINK_TESTS.assert(&HTTP_SINK_TAGS);
+        }
 
         assert_eq!(receiver.try_recv(), Ok(batch_status));
 
@@ -663,6 +667,7 @@ mod integration_tests {
             test_util::{onboarding_v2, BUCKET, ORG, TOKEN},
             InfluxDb2Settings,
         },
+        test_util::components::{self, HTTP_SINK_TAGS},
     };
     use chrono::Utc;
     use futures::stream;
@@ -708,7 +713,7 @@ mod integration_tests {
 
         let events = vec![Event::Log(event1), Event::Log(event2)];
 
-        sink.run(stream::iter(events)).await.unwrap();
+        components::run_sink(sink, stream::iter(events), &HTTP_SINK_TAGS).await;
 
         assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));
 

src/sinks/logdna.rs

@@ -296,6 +296,7 @@ mod tests {
     use crate::{
         config::SinkConfig,
         sinks::util::test::{build_test_server_status, load_sink},
+        test_util::components::{self, HTTP_SINK_TAGS},
         test_util::{next_addr, random_lines, trace_init},
     };
     use futures::{channel::mpsc, stream, StreamExt};
@@ -401,7 +402,11 @@ mod tests {
         }
         drop(batch);
 
+        components::init();
         sink.run(stream::iter(events)).await.unwrap();
+        if batch_status == BatchStatus::Delivered {
+            components::SINK_TESTS.assert(&HTTP_SINK_TAGS);
+        }
 
         assert_eq!(receiver.try_recv(), Ok(batch_status));