enhancement: batch events out metric in Pipeline (vectordotdev#8383)

* enhancement: batch events out metric in Pipeline

Signed-off-by: Toby Lawrence <[email protected]>

src/pipeline.rs

@@ -28,13 +28,23 @@ pub struct Pipeline {
     #[derivative(Debug = "ignore")]
     inlines: Vec<Box<dyn FunctionTransform>>,
     enqueued: VecDeque<Event>,
+    events_outstanding: usize,
 }
 
 impl Pipeline {
     fn try_flush(
         &mut self,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), <Self as Sink<Event>>::Error>> {
+        // We batch the updates to "events out" for efficiency, and do it here because
+        // it gives us a chance to allow the natural batching of `Pipeline` to kick in.
+        if self.events_outstanding > 0 {
+            emit!(EventOut {
+                count: self.events_outstanding
+            });
+            self.events_outstanding = 0;
+        }
+
         while let Some(event) = self.enqueued.pop_front() {
             match self.inner.poll_ready(cx) {
                 Poll::Pending => {
@@ -79,7 +89,8 @@ impl Sink<Event> for Pipeline {
     }
 
     fn start_send(mut self: Pin<&mut Self>, item: Event) -> Result<(), Self::Error> {
-        emit!(EventOut { count: 1 });
+        self.events_outstanding += 1;
+
         // Note how this gets **swapped** with `new_working_set` in the loop.
         // At the end of the loop, it will only contain finalized events.
         let mut working_set = vec![item];
@@ -142,6 +153,7 @@ impl Pipeline {
             // We ensure the buffer is sufficient that it is unlikely to require reallocations.
             // There is a possibility a component might blow this queue size.
             enqueued: VecDeque::with_capacity(10),
+            events_outstanding: 0,
         }
     }
 }