fix(instrument): Buffer counter underflowed (#23872) (#23973)

* fix(instrument): Buffer counter underflowed (#23872)

* Update code to be more clearly with to let

Co-authored-by: Bruce Guenter <[email protected]>

* format fix to meet format&clippy

---------

Co-authored-by: Bruce Guenter <[email protected]>
Co-authored-by: Pavlos Rontidis <[email protected]>

Author: 3 people

changelog.d/23872_buffer_counter_underflowed.fix.md

@@ -0,0 +1,3 @@
+Fix buffer counter underflowed, caused by the counter has not been updated(increase) timely when new event is coming.
+
+authors: sialais

lib/vector-buffers/src/topology/channel/sender.rs

@@ -202,8 +202,14 @@ impl<T: Bufferable> BufferSender<T> {
             .as_ref()
             .map(|_| (item.event_count(), item.size_of()));
 
-        let mut sent_to_base = true;
         let mut was_dropped = false;
+
+        if let Some(instrumentation) = self.instrumentation.as_ref()
+            && let Some((item_count, item_size)) = item_sizing
+        {
+            instrumentation
+                .increment_received_event_count_and_byte_size(item_count as u64, item_size as u64);
+        }
         match self.when_full {
             WhenFull::Block => self.base.send(item).await?,
             WhenFull::DropNewest => {
@@ -213,7 +219,7 @@ impl<T: Bufferable> BufferSender<T> {
             }
             WhenFull::Overflow => {
                 if let Some(item) = self.base.try_send(item).await? {
-                    sent_to_base = false;
+                    was_dropped = true;
                     self.overflow
                         .as_mut()
                         .unwrap_or_else(|| unreachable!("overflow must exist"))
@@ -223,30 +229,20 @@ impl<T: Bufferable> BufferSender<T> {
             }
         }
 
-        if (sent_to_base || was_dropped)
-            && let (Some(send_duration), Some(send_reference)) =
-                (self.send_duration.as_ref(), send_reference)
-        {
-            send_duration.emit(send_reference.elapsed());
-        }
-
         if let Some(instrumentation) = self.instrumentation.as_ref()
             && let Some((item_count, item_size)) = item_sizing
+            && was_dropped
         {
-            if sent_to_base {
-                instrumentation.increment_received_event_count_and_byte_size(
-                    item_count as u64,
-                    item_size as u64,
-                );
-            }
-
-            if was_dropped {
-                instrumentation.increment_dropped_event_count_and_byte_size(
-                    item_count as u64,
-                    item_size as u64,
-                    true,
-                );
-            }
+            instrumentation.increment_dropped_event_count_and_byte_size(
+                item_count as u64,
+                item_size as u64,
+                true,
+            );
+        }
+        if let Some(send_duration) = self.send_duration.as_ref()
+            && let Some(send_reference) = send_reference
+        {
+            send_duration.emit(send_reference.elapsed());
         }
 
         Ok(())