enhancement(observability): Add internal events for http source

src/internal_events/http.rs

@@ -0,0 +1,54 @@
+use super::InternalEvent;
+use metrics::counter;
+
+#[derive(Debug)]
+pub struct HTTPEventsReceived {
+    pub events_count: usize,
+    pub byte_size: usize,
+}
+
+impl InternalEvent for HTTPEventsReceived {
+    fn emit_logs(&self) {
+        trace!(
+            message = "sending events.",
+            events_count = %self.events_count,
+            byte_size = %self.byte_size,
+        );
+    }
+
+    fn emit_metrics(&self) {
+        counter!("events_processed", self.events_count as u64,
+            "component_kind" => "source",
+            "component_type" => "http",
+        );
+        counter!("bytes_processed", self.byte_size as u64,
+            "component_kind" => "source",
+            "component_type" => "http",
+        );
+    }
+}
+
+#[derive(Debug)]
+pub struct HTTPBadRequest<'a> {
+    pub error_code: u16,
+    pub error_message: &'a str,
+}
+
+impl<'a> InternalEvent for HTTPBadRequest<'a> {
+    fn emit_logs(&self) {
+        warn!(
+            message = "received bad request.",
+            code = %self.error_code,
+            message = %self.error_message,
+            rate_limit_secs = 10,
+        );
+    }
+
+    fn emit_metrics(&self) {
+        counter!(
+            "http_bad_requests", 1,
+            "component_kind" => "source",
+            "component_type" => "http",
+        );
+    }
+}

src/internal_events/mod.rs

@@ -3,6 +3,7 @@ mod aws_kinesis_streams;
 mod blackhole;
 mod elasticsearch;
 mod file;
+mod http;
 #[cfg(all(feature = "sources-journald", feature = "unix"))]
 mod journald;
 mod json;
@@ -30,6 +31,7 @@ pub use self::aws_kinesis_streams::*;
 pub use self::blackhole::*;
 pub use self::elasticsearch::*;
 pub use self::file::*;
+pub use self::http::*;
 #[cfg(all(feature = "sources-journald", feature = "unix"))]
 pub(crate) use self::journald::*;
 pub use self::json::*;

src/sources/util/http.rs

@@ -1,8 +1,10 @@
-use crate::event::Event;
 use crate::{
+    event::Event,
+    internal_events::{HTTPBadRequest, HTTPEventsReceived},
     shutdown::ShutdownSignal,
     tls::{MaybeTlsSettings, TlsConfig},
 };
+use bytes05::Bytes;
 use futures::{
     compat::{AsyncRead01CompatExt, Future01CompatExt, Stream01CompatExt},
     FutureExt, TryFutureExt, TryStreamExt,
@@ -74,15 +76,20 @@ pub trait HttpSource: Clone + Send + Sync + 'static {
             .and(warp::path::end())
             .and(warp::header::headers_cloned())
             .and(warp::body::bytes())
-            .and_then(move |headers: HeaderMap, body| {
+            .and_then(move |headers: HeaderMap, body: Bytes| {
                 info!("Handling http request: {:?}", headers);
 
                 let this = self.clone();
                 let out = out.clone();
 
                 async move {
+                    let body_size = body.len();
                     match this.build_event(body, headers) {
                         Ok(events) => {
+                            emit!(HTTPEventsReceived {
+                                events_count: events.len(),
+                                byte_size: body_size,
+                            });
                             out.send_all(futures01::stream::iter_ok(events))
                                 .compat()
                                 .map_err(move |e: mpsc::SendError<Event>| {
@@ -95,7 +102,13 @@ pub trait HttpSource: Clone + Send + Sync + 'static {
                                 .map_ok(|_| warp::reply())
                                 .await
                         }
-                        Err(err) => Err(warp::reject::custom(err)),
+                        Err(err) => {
+                            emit!(HTTPBadRequest {
+                                error_code: err.code,
+                                error_message: err.message.as_str(),
+                            });
+                            Err(warp::reject::custom(err))
+                        }
                     }
                 }
             });