chore(connlib): record GRO batch-size as histogram metric (#8875)

This will allow us to test, how many batches we typically read from the
UDP socket in a single syscall. Knowing that should help in profiling
#8874.

rust/Cargo.lock

@@ -6487,6 +6487,7 @@ dependencies = [
  "gat-lending-iterator",
  "ip-packet",
  "lockfree-object-pool",
+ "opentelemetry",
  "parking_lot",
  "quinn-udp",
  "socket2",

rust/socket-factory/Cargo.toml

@@ -12,6 +12,7 @@ firezone-logging = { workspace = true }
 gat-lending-iterator = { workspace = true }
 ip-packet = { workspace = true }
 lockfree-object-pool = { workspace = true }
+opentelemetry = { workspace = true, features = ["metrics"] }
 parking_lot = { workspace = true }
 quinn-udp = { workspace = true }
 socket2 = { workspace = true }

rust/socket-factory/src/lib.rs

@@ -3,6 +3,7 @@ use bytes::Buf as _;
 use firezone_logging::err_with_src;
 use gat_lending_iterator::LendingIterator;
 use ip_packet::Ecn;
+use opentelemetry::KeyValue;
 use parking_lot::Mutex;
 use quinn_udp::{EcnCodepoint, Transmit};
 use std::collections::HashMap;
@@ -153,6 +154,7 @@ pub struct UdpSocket {
     /// A buffer pool for batches of incoming UDP packets.
     buffer_pool: Arc<lockfree_object_pool::MutexObjectPool<Vec<u8>>>,
 
+    gro_batch_histogram: opentelemetry::metrics::Histogram<u64>,
     port: u16,
 }
 
@@ -170,6 +172,14 @@ impl UdpSocket {
                 || vec![0u8; u16::MAX as usize],
                 |_| {},
             )),
+            gro_batch_histogram: opentelemetry::global::meter("connlib")
+                .u64_histogram("system.network.packets.batch_count")
+                .with_description(
+                    "How many batches of packets we have processed in a single syscall.",
+                )
+                .with_unit("{batches}")
+                .with_boundaries((1..32_u64).map(|i| i as f64).collect())
+                .init(),
         })
     }
 
@@ -248,7 +258,15 @@ impl UdpSocket {
                 state.recv(socket, &mut bufs, &mut meta)
             };
 
-            if let Ok(_len) = inner.try_io(Interest::READABLE, recv) {
+            if let Ok(len) = inner.try_io(Interest::READABLE, recv) {
+                self.gro_batch_histogram.record(
+                    len as u64,
+                    &[
+                        KeyValue::new("network.transport", "udp"),
+                        KeyValue::new("network.io.direction", "receive"),
+                    ],
+                );
+
                 // Note: We don't need to use `len` here because the iterator will stop once it encounteres `meta.len == 0`.
 
                 return Poll::Ready(Ok(DatagramSegmentIter::new(bufs, meta, *port)));