Rate-limit logging of unmatched packets

Log the first unmatched packet per source (WARN for an unconfigured source address, INFO for an unknown Your Discriminator) and the repeats at DEBUG until a packet matches again. Removes the duplicate message in rx_packet.

src/bin/rust-bfdd.rs

@@ -5,7 +5,7 @@ use rust_bfd::events::{EventMessageSink, ScriptHookSink, StateChangeEvent};
 use rust_bfd::{bfd::*, events::EventMessage};
 use rust_bfd::{set_ttl_or_hop_limit, BfdSessionStats};
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     error::Error,
     fmt::Display,
     io::ErrorKind,
@@ -999,6 +999,8 @@ struct Bfdd {
     peers: Vec<ResolvedPeer>,
     sessions_by_ip: HashMap<(IpAddr, IpAddr), BfdDiscriminator>,
     sessions_by_discr: HashMap<BfdDiscriminator, BfdSessionHandle>,
+    // Sources we have already warned about unmatched packets from; cleared per source once a packet matches again
+    unmatched_sources: std::sync::Mutex<HashSet<IpAddr>>,
 
     control_socket: Arc<UdpSocket>,
 
@@ -1039,6 +1041,7 @@ impl Bfdd {
             peers,
             sessions_by_ip,
             sessions_by_discr,
+            unmatched_sources: std::sync::Mutex::new(HashSet::new()),
             control_socket,
             event_dispatcher,
         })
@@ -1090,13 +1093,38 @@ impl Bfdd {
                 );
                 discr
             } else {
-                warn!("Unable to match packet to session with {}", addr.ip());
+                self.log_unmatched(
+                    addr.ip(),
+                    log::Level::Warn,
+                    "no session configured for this address",
+                );
                 return None;
             }
         } else {
             &packet.your_disc
         };
-        self.sessions_by_discr.get(our_disc)
+        let session = self.sessions_by_discr.get(our_disc);
+        if session.is_some() {
+            self.unmatched_sources.lock().unwrap().remove(&addr.ip());
+        } else {
+            self.log_unmatched(
+                addr.ip(),
+                log::Level::Info,
+                &format!("unknown Your Discriminator {our_disc}"),
+            );
+        }
+        session
+    }
+    /// Log the first unmatched packet from a source at `level`, then at debug level until one matches again.
+    fn log_unmatched(&self, source: IpAddr, level: log::Level, reason: &str) {
+        if self.unmatched_sources.lock().unwrap().insert(source) {
+            log::log!(
+                level,
+                "Unable to match packet from {source}: {reason} (repeats are logged at debug level)"
+            );
+        } else {
+            debug!("Unable to match packet from {source}: {reason}");
+        }
     }
     async fn rx_packet(&self, addr: SocketAddr, packet: BfdPacket) {
         if let Some(session) = self.session_for_packet(&addr, &packet) {
@@ -1105,8 +1133,6 @@ impl Bfdd {
                 .send(SessionControlCommand::RxPacket(packet))
                 .await
                 .unwrap();
-        } else {
-            warn!("Unable to match packet to session with");
         }
     }
     async fn run(mut self) -> io::Result<()> {