[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;
@@ -791,16 +792,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,23 +811,20 @@ 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())
     }
 }
 

dpd-client/tests/integration_tests/counters.rs

@@ -11,7 +11,9 @@
 
 use std::net::Ipv4Addr;
 use std::sync::Arc;
+use std::time::Duration;
 
+use anyhow::bail;
 use packet::Endpoint;
 
 use crate::integration_tests::common;
@@ -22,6 +24,13 @@ use dpd_client::ClientInfo;
 use dpd_client::types::Ipv4Entry;
 use dpd_client::types::Ipv6Entry;
 
+// 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;
+
 /// Awaits an increase in a given counter's value from `baseline` to
 /// `baseline + increment`.
 ///
@@ -33,32 +42,36 @@ pub async fn check_counter_incremented(
     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;
 
     // Briefly delay before reading the counter each time, as there is
     // occasionally some lag between when we see the counter set in the
     // simulator and when that updated value is available to the SDE.  To
     // avoid a long pointless delay here, we try multiple times with a short
     // sleep rather than once with a long sleep.
-    for _ in 0..20 {
-        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
-        new_value =
-            switch.get_counter(counter_name, client_name).await.unwrap();
+    for _ in 0..COUNTER_POLL_ATTEMPTS {
+        tokio::time::sleep(COUNTER_POLL_INTERVAL).await;
+        new_value = switch.get_counter(counter_name, client_name).await?;
 
-        if new_value == baseline + expected_increment {
+        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.
-    Err(anyhow::anyhow!(
-        "Counter '{}' expected to increase by {} (from {} to {}), but only reached {}",
-        counter_name,
-        expected_increment,
-        baseline,
-        baseline + expected_increment,
-        new_value
-    ))
+    bail!(
+        "{incr_expect}, but reached {new_value} after \
+         {COUNTER_POLL_ATTEMPTS} polls"
+    )
 }
 
 /// Process a single packet, asserting that a given counter is incremented by
@@ -68,10 +81,11 @@ pub async fn one_counted_test(
     packet: TestPacket,
     packet_after: Option<TestPacket>,
     counter: &str,
+    client_name: Option<&str>,
 ) -> anyhow::Result<bool> {
-    let old = switch.get_counter(counter, None).await?;
+    let old = switch.get_counter(counter, client_name).await?;
     switch.packet_test(vec![packet], packet_after.into_iter().collect())?;
-    check_counter_incremented(switch, counter, old, 1, None)
+    check_counter_incremented(switch, counter, old, 1, client_name)
         .await
         .map(|new| old + 1 == new)
 }
@@ -89,6 +103,7 @@ async fn one_drop_test(
         TestPacket { packet: Arc::new(packet), port },
         None,
         counter,
+        None,
     )
     .await
 }