[Agent] Add npb monitoring

deepflowio · Jun 6, 2024 · 10cd991 · 10cd991
1 parent c1c2c7b
commit 10cd991
Show file tree

Hide file tree

Showing 3 changed files with 37 additions and 20 deletions.
diff --git a/agent/src/config/handler.rs b/agent/src/config/handler.rs
@@ -252,6 +252,7 @@ pub struct NpbConfig {
     pub vlan_mode: trident::VlanMode,
     pub socket_type: trident::SocketType,
     pub ignore_overlay_vlan: bool,
+    pub queue_size: usize,
 }
 
 impl Default for NpbConfig {
@@ -1513,6 +1514,7 @@ impl TryFrom<(Config, RuntimeConfig)> for ModuleConfig {
                 vlan_mode: conf.npb_vlan_mode,
                 dedup_enabled: conf.npb_dedup_enabled,
                 socket_type: conf.npb_socket_type,
+                queue_size: conf.yaml_config.collector_sender_queue_size,
             },
             collector: CollectorConfig {
                 enabled: conf.collector_enabled,

diff --git a/agent/src/handler/npb.rs b/agent/src/handler/npb.rs
@@ -38,7 +38,7 @@ use crate::common::{
 };
 use crate::config::NpbConfig;
 use crate::sender::npb_sender::{NpbArpTable, NpbPacketSender};
-use crate::utils::stats::{self, StatsOption};
+use crate::utils::stats::{self, QueueStats, StatsOption};
 use npb_handler::{NpbHandler, NpbHandlerCounter, NpbHeader, StatsNpbHandlerCounter, NOT_SUPPORT};
 use public::{
     counter::Countable,
@@ -208,8 +208,17 @@ impl NpbBuilder {
         self.stop();
         self.npb_packet_sender = None;
 
-        let (sender, receiver, _) =
-            bounded_with_debug(4096, "1-packet-to-npb-sender", queue_debugger);
+        let queue_name = "1-packet-to-npb-sender";
+        let (sender, receiver, counter) =
+            bounded_with_debug(config.queue_size, queue_name, queue_debugger);
+        self.stats_collector.register_countable(
+            &QueueStats {
+                id: self.id,
+                module: queue_name,
+            },
+            Countable::Owned(Box::new(counter)),
+        );
+
         let npb_packet_sender = Arc::new(NpbPacketSender::new(
             self.id,
             receiver,
@@ -245,8 +254,16 @@ impl NpbBuilder {
         arp: Arc<NpbArpTable>,
         stats_collector: Arc<stats::Collector>,
     ) -> Box<Self> {
-        let (sender, receiver, _) =
-            bounded_with_debug(4096, "1-packet-to-npb-sender", queue_debugger);
+        let queue_name = "1-packet-to-npb-sender";
+        let (sender, receiver, counter) =
+            bounded_with_debug(config.queue_size, queue_name, queue_debugger);
+        stats_collector.register_countable(
+            &QueueStats {
+                id,
+                module: queue_name,
+            },
+            Countable::Owned(Box::new(counter)),
+        );
 
         let builder = Box::new(Self {
             id,

diff --git a/agent/src/sender/npb_sender.rs b/agent/src/sender/npb_sender.rs
@@ -481,13 +481,7 @@ impl NpbSender {
 pub struct NpbSenderCounter {
     pub tx: AtomicUsize,
     pub tx_bytes: AtomicUsize,
-}
-
-impl NpbSenderCounter {
-    fn reset(&self) {
-        self.tx.store(0, Ordering::Relaxed);
-        self.tx_bytes.store(0, Ordering::Relaxed);
-    }
+    pub tx_dropped: AtomicUsize,
 }
 
 pub struct StatsNpbSenderCounter(Weak<NpbSenderCounter>);
@@ -500,18 +494,21 @@ impl OwnedCountable for StatsNpbSenderCounter {
     fn get_counters(&self) -> Vec<public::counter::Counter> {
         match self.0.upgrade() {
             Some(x) => {
-                let (tx, tx_bytes) = (
-                    x.tx.load(Ordering::Relaxed) as u64,
-                    x.tx_bytes.load(Ordering::Relaxed) as u64,
-                );
-                x.reset();
-
                 vec![
-                    ("tx", CounterType::Counted, CounterValue::Unsigned(tx)),
+                    (
+                        "tx",
+                        CounterType::Counted,
+                        CounterValue::Unsigned(x.tx.swap(0, Ordering::Relaxed) as u64),
+                    ),
                     (
                         "tx_bytes",
                         CounterType::Counted,
-                        CounterValue::Unsigned(tx_bytes),
+                        CounterValue::Unsigned(x.tx_bytes.swap(0, Ordering::Relaxed) as u64),
+                    ),
+                    (
+                        "tx_dropped",
+                        CounterType::Counted,
+                        CounterValue::Unsigned(x.tx_dropped.swap(0, Ordering::Relaxed) as u64),
                     ),
                 ]
             }
@@ -907,6 +904,7 @@ impl NpbConnectionPool {
         let bytes = packet.len();
         let ret = self.send_to(timestamp, underlay_l2_opt_size, packet);
         if ret.is_err() {
+            self.counter.tx_dropped.fetch_add(1, Ordering::Relaxed);
             return ret;
         }
         self.counter.tx.fetch_add(1, Ordering::Relaxed);