[p4, sidecar] fix underlay/external mcast counter ops

common/src/counters.rs

@@ -13,6 +13,10 @@ pub use dpd_types::counters::{
     FecRSCounters, PcsCounters, RMonCounters, RMonCountersAll,
 };
 
+/// Per-port counter label for the userspace service port, which has no link
+/// of its own on the current switch.
+pub const SERVICE_PORT_LABEL: &str = "service";
+
 /// sidecar.p4 defines the following set of indirect counters.
 #[derive(
     Debug,

dpd-client/tests/integration_tests/common.rs

@@ -14,6 +14,7 @@ use std::{fmt, thread};
 
 use anyhow::Context;
 use anyhow::anyhow;
+use anyhow::bail;
 use oxnet::Ipv4Net;
 use oxnet::Ipv6Net;
 use slog::Drain;
@@ -45,6 +46,13 @@ pub const ADMIN_LOCAL_MULTICAST_PREFIX: u16 = 0xFF04;
 // packets before notifying us that any are ready.
 const TEST_PCAP_TIMEOUT_MS: i32 = 1;
 
+// Interval and attempt count for polling a counter until it reaches an
+// expected value.
+//
+// Note: counter updates from the model can lag the packet itself.
+const COUNTER_POLL_INTERVAL: Duration = Duration::from_millis(100);
+const COUNTER_POLL_ATTEMPTS: usize = 20;
+
 // Physical port number
 #[derive(Clone, Copy, PartialOrd, Ord, Hash, PartialEq, Eq)]
 pub struct PhysPort(pub u16);
@@ -791,16 +799,10 @@ impl Switch {
             }
         }
 
-        match errors.len() {
-            0 => Ok(()),
-            _ => Err(anyhow!(
-                errors
-                    .iter()
-                    .map(|e| e.to_string())
-                    .collect::<Vec<String>>()
-                    .join(", ")
-            )),
+        if !errors.is_empty() {
+            bail!("{}", errors.join(", "));
         }
+        Ok(())
     }
 
     /// If no client_name is given, this returns the number of packets dropped
@@ -816,26 +818,58 @@ impl Switch {
         counter: &str,
         client_name: Option<&str>,
     ) -> anyhow::Result<u64> {
-        let client_name = if let Some(client_name) = client_name {
-            client_name.to_string()
-        } else {
-            "ingress_drop_reason".to_string()
-        };
+        let client_name = client_name.unwrap_or("ingress_drop_reason");
 
-        self.client
-            .counter_get(&client_name, true)
+        let entries = self
+            .client
+            .counter_get(client_name, true)
             .await
-            .map_err(|e| anyhow::anyhow!("failed to fetch counters: {e:?}"))
-            .and_then(|entries| {
-                entries
-                    .iter()
-                    .find(|e| e.keys.get("label").unwrap().as_str() == counter)
-                    .map(|e| e.data.pkts.unwrap())
-                    .ok_or(anyhow::anyhow!("no such counter: {counter}"))
-            })
+            .map_err(|e| anyhow!("failed to fetch counters: {e:?}"))?;
+        let Some(entry) = entries
+            .iter()
+            .find(|e| e.keys.get("label").unwrap().as_str() == counter)
+        else {
+            bail!("no such counter: {counter} in {client_name}");
+        };
+        Ok(entry.data.pkts.unwrap())
     }
 }
 
+pub async fn check_counter_incremented(
+    switch: &Switch,
+    counter_name: &str,
+    baseline: u64,
+    expected_increment: u64,
+    client_name: Option<&str>,
+) -> anyhow::Result<u64> {
+    let expected = baseline + expected_increment;
+    let table = client_name.unwrap_or("ingress_drop_reason");
+    let incr_expect = format!(
+        "counter '{counter_name}' in {table} expected to increase by \
+         {expected_increment} ({baseline} -> {expected})"
+    );
+    let mut new_value = 0;
+
+    // Poll for the counter value (with timeout)
+    for _i in 0..COUNTER_POLL_ATTEMPTS {
+        thread::sleep(COUNTER_POLL_INTERVAL);
+        new_value = switch.get_counter(counter_name, client_name).await?;
+
+        if new_value == expected {
+            return Ok(new_value);
+        }
+        if new_value > expected {
+            bail!("{incr_expect}, but overshot to {new_value}");
+        }
+    }
+
+    // Counter didn't increment as expected
+    bail!(
+        "{incr_expect}, but reached {new_value} after \
+         {COUNTER_POLL_ATTEMPTS} polls"
+    )
+}
+
 // Construct a single TCP packet with an optional payload
 pub fn gen_tcp_packet_loaded(
     src: Endpoint,
@@ -1516,5 +1550,6 @@ pub mod prelude {
     pub use super::Switch;
     pub use super::TestPacket;
     pub use super::TestResult;
+    pub use super::check_counter_incremented;
     pub use super::get_switch;
 }