[Backport release-mainnet-1.1.0-rc #4327] chore(libp2p): reduce log noise via debug demotion and periodic summary (#4371)

chore(libp2p): reduce log noise via debug demotion and periodic summary (#4327)

* chore(libp2p): reduce log noise via debug demotion and periodic summary

- demote per-event WARNs (dial failures, Kademlia NotFound, auth handshake
  failures, slow gossipsub peers, DHT internal disagreement, direct-message
  failures, gossip publish failures, DHT save concurrency, bootstrap
  duplicate-start) from WARN/ERROR to DEBUG; production WARN volume drops
  from ~5800/hr to <=1/min per node
- add network::log_summary with 16 atomic counters and an idempotent
  spawn_summary_task that emits one aggregated WARN line per minute when
  any counter is non-zero, format is alphabetical and emits only non-zero
  values for stable parsing
- counters increment at every demoted site before the debug call so totals
  survive even when DEBUG is filtered out
- wire counters across libp2p-networking, hotshot, and hotshot-task-impls;
  add hotshot-libp2p-networking workspace dep to task-impls
- remove libp2p=off from .env now that crate-level log gating is unnecessary

* refactor(libp2p): emit log summary as structured tracing fields

- replace free-form "name=value name=value" message with a warn! event
  carrying one named tracing field per counter (auth_failures, dial_failures,
  ...), plus interval_seconds, so structured backends like Datadog index and
  filter on each counter without regex
- drain counters atomically into a Snapshot struct; emit only when
  any counter is non-zero
- always emit all counters, including zero values, so Datadog field schemas
  stay stable across intervals
- add tracing-test dev-dep and a traced_test that asserts named fields
  appear in the emitted event

* chore(libp2p): demote log summary to info, omit zero counters

- emit the libp2p periodic summary at info level; a heartbeat of normal
  p2p churn is informational, not warn-worthy
- format only non-zero counters; zero values were padding the line without
  helping operators or Datadog filters
- replace Snapshot struct with a flat COUNTERS table; drain_and_format
  iterates once, returns None when everything is zero
- richer per-counter metrics belong in the existing Prometheus
  Libp2pMetricsValue path, not in this log line

* refactor(libp2p): shrink log_summary module

- collapse 16 static counters + COUNTERS table into a single counters! macro
- drop plan-traceability test comments and the per-counter unit tests; keep
  one smoke test for emit behavior and one for concurrent correctness
- trim docstring to a single paragraph
- use thread::scope in the concurrency test, no manual join

* chore(libp2p): drop unused return value of spawn_summary_task

Also drop obvious comments restating function names.

* refactor(libp2p): replace static counters with LogEvent::record() API

- log_summary: 16 pub statics become one LogEvent enum + parallel
  NAMES/COUNTERS arrays indexed by `event as usize`; record is a method
  on LogEvent
- call sites change from
    log_summary::X.fetch_add(1, Ordering::Relaxed);
  to
    LogEvent::X.record();
  one import per file, one reach per call

* chore(libp2p): drop log_summary header

* fix(libp2p): only record DhtDisagreementGivenUp on terminal failure

The counter (and the misleading "Giving up because out of retries" log)
were firing unconditionally, including for retries that succeeded. Move
both into the else branch where retry_count == 0.

Also switch the SPAWNED CAS from Relaxed/Relaxed to AcqRel/Acquire, the
idiomatic ordering for a "do once" guard.

(cherry picked from commit be6ff77)

Author: sveitser

.env

@@ -15,7 +15,7 @@
 #   sequencer service.
 #
 
-RUST_LOG=info,libp2p=off,cliquenet=error
+RUST_LOG=info,cliquenet=error
 RUST_LOG_FORMAT=full
 
 # Parallelism config

Cargo.lock

@@ -39,6 +39,7 @@ use hotshot_libp2p_networking::{
             record::{Namespace, RecordKey, RecordValue},
             store::persistent::DhtPersistentStorage,
         },
+        log_summary::LogEvent,
         spawn_network_node,
         transport::construct_auth_message,
     },
@@ -74,7 +75,7 @@ use tokio::{
     },
     time::sleep,
 };
-use tracing::{error, info, instrument, trace, warn};
+use tracing::{debug, error, info, instrument, trace, warn};
 
 use crate::{BroadcastDelay, EpochMembershipCoordinator};
 
@@ -609,7 +610,8 @@ impl<T: NodeType> Libp2pNetwork<T> {
                 if latest_seen_view.load(Ordering::Relaxed) + THRESHOLD <= *view_number {
                     // look up
                     if let Err(err) = handle.lookup_node(&pk, dht_timeout).await {
-                        warn!("Failed to perform lookup for key {pk}: {err}");
+                        LogEvent::DhtLookupFailure.record();
+                        debug!("Failed to perform lookup for key {pk}: {err}");
                     };
                 }
             }

crates/hotshot/hotshot/src/traits/networking/libp2p_network.rs

@@ -38,6 +38,7 @@ tracing-subscriber = { workspace = true }
 
 [dev-dependencies]
 hotshot-example-types = { workspace = true }
+tracing-test = { workspace = true }
 
 [lints]
 workspace = true

crates/hotshot/libp2p-networking/Cargo.toml

@@ -56,7 +56,7 @@ impl DHTBootstrapTask {
                         break;
                     },
                     Some(InputEvent::StartBootstrap) => {
-                        tracing::warn!("Trying to start bootstrap that's already in progress");
+                        tracing::debug!("Trying to start bootstrap that's already in progress");
                         continue;
                     },
                     None => {

crates/hotshot/libp2p-networking/src/network/behaviours/dht/bootstrap.rs

@@ -42,7 +42,7 @@ lazy_static! {
 }
 
 use super::exponential_backoff::ExponentialBackoff;
-use crate::network::{ClientRequest, NetworkEvent};
+use crate::network::{ClientRequest, NetworkEvent, log_summary::LogEvent};
 
 /// Behaviour wrapping libp2p's kademlia
 /// included:
@@ -184,7 +184,9 @@ impl<K: SignatureKey + 'static, D: DhtPersistentStorage> DHTBehaviour<K, D> {
             // The key already exists in the cache, send the value to all channels
             for chan in chans {
                 if chan.send(entry.value.clone()).is_err() {
-                    warn!("Get DHT: channel closed before get record request result could be sent");
+                    debug!(
+                        "Get DHT: channel closed before get record request result could be sent"
+                    );
                 }
             }
         } else {
@@ -280,7 +282,8 @@ impl<K: SignatureKey + 'static, D: DhtPersistentStorage> DHTBehaviour<K, D> {
                     },
                 },
                 Err(err) => {
-                    warn!("Error in Kademlia query: {err:?}");
+                    LogEvent::DhtKadQueryError.record();
+                    debug!("Error in Kademlia query: {err:?}");
                     false
                 },
             },
@@ -323,7 +326,7 @@ impl<K: SignatureKey + 'static, D: DhtPersistentStorage> DHTBehaviour<K, D> {
                     // Send the record to all channels that are still open
                     for n in notify {
                         if n.send(record.value.clone()).is_err() {
-                            warn!(
+                            debug!(
                                 "Get DHT: channel closed before get record request result could \
                                  be sent"
                             );
@@ -334,25 +337,23 @@ impl<K: SignatureKey + 'static, D: DhtPersistentStorage> DHTBehaviour<K, D> {
                 }
             }
             // disagreement => query more nodes
-            else {
-                // there is some internal disagreement or not enough nodes returned
-                // Initiate new query that hits more replicas
-                if retry_count > 0 {
-                    let new_retry_count = retry_count - 1;
-                    warn!(
-                        "Get DHT: Internal disagreement for get dht request {progress:?}! \
-                         requerying with more nodes. {new_retry_count:?} retries left"
-                    );
-                    self.retry_get(KadGetQuery {
-                        backoff,
-                        progress: DHTProgress::NotStarted,
-                        notify,
-                        key,
-                        retry_count: new_retry_count,
-                        records: Vec::new(),
-                    });
-                }
-                warn!(
+            else if retry_count > 0 {
+                let new_retry_count = retry_count - 1;
+                debug!(
+                    "Get DHT: Internal disagreement for get dht request {progress:?}! requerying \
+                     with more nodes. {new_retry_count:?} retries left"
+                );
+                self.retry_get(KadGetQuery {
+                    backoff,
+                    progress: DHTProgress::NotStarted,
+                    notify,
+                    key,
+                    retry_count: new_retry_count,
+                    records: Vec::new(),
+                });
+            } else {
+                LogEvent::DhtDisagreementGivenUp.record();
+                debug!(
                     "Get DHT: Internal disagreement for get dht request {progress:?}! Giving up \
                      because out of retries. "
                 );
@@ -438,7 +439,8 @@ impl<K: SignatureKey + 'static, D: DhtPersistentStorage> DHTBehaviour<K, D> {
                     if let Some(chan) = self.in_progress_get_closest_peers.remove(&query_id) {
                         let _: Result<_, _> = chan.send(());
                     };
-                    warn!("Failed to get closest peers: {e:?}");
+                    LogEvent::DhtClosestPeersFailure.record();
+                    debug!("Failed to get closest peers: {e:?}");
                 },
             },
             KademliaEvent::OutboundQueryProgressed {

crates/hotshot/libp2p-networking/src/network/behaviours/dht/mod.rs

@@ -243,9 +243,9 @@ impl<R: RecordStore, D: DhtPersistentStorage> PersistentStore<R, D> {
             semaphore: Arc::new(Semaphore::new(1)),
         };
 
-        // Try to restore the DHT from the persistent store. If it fails, warn and start with an empty store
+        // Try to restore the DHT from the persistent store. If it fails, log and start with an empty store
         if let Err(err) = store.restore_from_persistent_storage().await {
-            warn!(
+            debug!(
                 "Failed to restore DHT from persistent storage: {err}. Starting with empty store",
             );
         }
@@ -258,9 +258,9 @@ impl<R: RecordStore, D: DhtPersistentStorage> PersistentStore<R, D> {
     ///
     /// Returns `true` if the DHT was saved, `false` otherwise.
     fn try_save_to_persistent_storage(&mut self) -> bool {
-        // Try to acquire the semaphore, warning if another save operation is already in progress
+        // Try to acquire the semaphore; if another save operation is already in progress, skip.
         let Ok(permit) = Arc::clone(&self.semaphore).try_acquire_owned() else {
-            warn!(
+            debug!(
                 "Skipping DHT save to persistent storage - another save operation is already in \
                  progress"
             );

crates/hotshot/libp2p-networking/src/network/behaviours/dht/store/persistent.rs

@@ -9,10 +9,10 @@ use std::collections::HashMap;
 use libp2p::request_response::{Event, Message, OutboundRequestId, ResponseChannel};
 use libp2p_identity::PeerId;
 use tokio::{spawn, sync::mpsc::UnboundedSender, time::sleep};
-use tracing::{debug, error, warn};
+use tracing::debug;
 
 use super::exponential_backoff::ExponentialBackoff;
-use crate::network::{ClientRequest, NetworkEvent};
+use crate::network::{ClientRequest, NetworkEvent, log_summary::LogEvent};
 
 /// Request to direct message a peert
 #[derive(Debug)]
@@ -58,7 +58,8 @@ impl DMBehaviour {
                 error,
                 connection_id: _,
             } => {
-                error!("Inbound message failure from {:?}: {:?}", peer, error);
+                LogEvent::DirectMessageInboundFailure.record();
+                debug!("Inbound message failure from {:?}: {:?}", peer, error);
                 None
             },
             Event::OutboundFailure {
@@ -67,7 +68,8 @@ impl DMBehaviour {
                 error,
                 connection_id: _,
             } => {
-                warn!("Outbound message failure to {:?}: {:?}", peer, error);
+                LogEvent::DirectMessageOutboundFailure.record();
+                debug!("Outbound message failure to {:?}: {:?}", peer, error);
                 if let Some(mut req) = self.in_progress_rr.remove(&request_id) {
                     if req.retry_count == 0 {
                         return None;
@@ -106,7 +108,7 @@ impl DMBehaviour {
                         debug!("Received direct response {:?}", msg);
                         Some(NetworkEvent::DirectResponse(msg, req.peer_id))
                     } else {
-                        warn!("Received response for unknown request id {:?}", request_id);
+                        debug!("Received response for unknown request id {:?}", request_id);
                         None
                     }
                 },

crates/hotshot/libp2p-networking/src/network/behaviours/direct_message.rs

@@ -23,6 +23,7 @@ use super::{
         validated::ValidatedStore,
     },
     cbor,
+    log_summary::LogEvent,
 };
 
 /// Overarching network behaviour performing:
@@ -98,7 +99,8 @@ impl<K: SignatureKey + 'static, D: DhtPersistentStorage> NetworkDef<K, D> {
     /// Publish a given gossip
     pub fn publish_gossip(&mut self, topic: IdentTopic, contents: Vec<u8>) {
         if let Err(e) = self.gossipsub.publish(topic, contents) {
-            tracing::warn!("Failed to publish gossip message. Error: {:?}", e);
+            LogEvent::GossipPublishFailure.record();
+            tracing::debug!("Failed to publish gossip message. Error: {:?}", e);
         }
     }
     /// Subscribe to a given topic

crates/hotshot/libp2p-networking/src/network/def.rs

@@ -0,0 +1,148 @@
+use std::{
+    sync::atomic::{AtomicBool, AtomicU64, Ordering},
+    time::Duration,
+};
+
+use tokio::time::interval;
+use tracing::info;
+
+pub const SUMMARY_INTERVAL: Duration = Duration::from_secs(60);
+
+macro_rules! events {
+    ($($variant:ident => $name:literal),* $(,)?) => {
+        #[derive(Clone, Copy)]
+        #[repr(usize)]
+        pub enum LogEvent {
+            $($variant),*
+        }
+
+        const NAMES: &[&str] = &[$($name),*];
+        static COUNTERS: [AtomicU64; NAMES.len()] =
+            [const { AtomicU64::new(0) }; NAMES.len()];
+    };
+}
+
+events! {
+    AuthFailure => "auth_failures",
+    AuthHandshakeTimeout => "auth_handshake_timeouts",
+    DhtClosestPeersFailure => "dht_closest_peers_failures",
+    DhtDisagreementGivenUp => "dht_disagreements_given_up",
+    DhtKadQueryError => "dht_kad_query_errors",
+    DhtLookupFailure => "dht_lookup_failures",
+    DialFailure => "dial_failures",
+    DirectMessageInboundFailure => "direct_message_inbound_failures",
+    DirectMessageOutboundFailure => "direct_message_outbound_failures",
+    GossipPublishFailure => "gossip_publish_failures",
+    GossipsubNotSupported => "gossipsub_not_supported",
+    GossipsubSlowPeer => "gossipsub_slow_peer",
+    IncomingConnError => "incoming_conn_errors",
+    ListenerError => "listener_errors",
+    NetworkSendFailure => "network_send_failures",
+    VerifyFailure => "verify_failures",
+}
+
+impl LogEvent {
+    pub fn record(self) {
+        COUNTERS[self as usize].fetch_add(1, Ordering::Relaxed);
+    }
+}
+
+fn drain_and_format() -> Option<String> {
+    let parts: Vec<String> = COUNTERS
+        .iter()
+        .zip(NAMES.iter())
+        .filter_map(|(counter, name)| {
+            let value = counter.swap(0, Ordering::Relaxed);
+            (value != 0).then(|| format!("{name}={value}"))
+        })
+        .collect();
+    (!parts.is_empty()).then(|| parts.join(" "))
+}
+
+fn emit_summary() {
+    if let Some(body) = drain_and_format() {
+        info!("libp2p {}s summary: {body}", SUMMARY_INTERVAL.as_secs());
+    }
+}
+
+static SPAWNED: AtomicBool = AtomicBool::new(false);
+
+pub fn spawn_summary_task() {
+    if SPAWNED
+        .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+        .is_err()
+    {
+        return;
+    }
+    tokio::spawn(async move {
+        let mut ticker = interval(SUMMARY_INTERVAL);
+        ticker.tick().await; // skip immediate first tick
+        loop {
+            ticker.tick().await;
+            emit_summary();
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{Mutex, MutexGuard, OnceLock, atomic::Ordering};
+
+    use tracing_test::traced_test;
+
+    use super::{COUNTERS, LogEvent, drain_and_format, emit_summary};
+
+    fn test_lock() -> MutexGuard<'static, ()> {
+        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
+        LOCK.get_or_init(|| Mutex::new(()))
+            .lock()
+            .unwrap_or_else(|p| p.into_inner())
+    }
+
+    fn reset() {
+        for c in COUNTERS.iter() {
+            c.store(0, Ordering::Relaxed);
+        }
+    }
+
+    #[test]
+    #[traced_test]
+    fn emits_only_nonzero_and_skips_when_idle() {
+        let _g = test_lock();
+        reset();
+        emit_summary();
+        assert!(!logs_contain("libp2p"));
+
+        for _ in 0..3 {
+            LogEvent::DialFailure.record();
+        }
+        for _ in 0..5 {
+            LogEvent::AuthFailure.record();
+        }
+        emit_summary();
+        assert!(logs_contain("libp2p 60s summary:"));
+        assert!(logs_contain("auth_failures=5"));
+        assert!(logs_contain("dial_failures=3"));
+        assert!(!logs_contain("verify_failures"));
+        assert!(drain_and_format().is_none());
+    }
+
+    #[test]
+    fn concurrent_increments_are_not_lost() {
+        let _g = test_lock();
+        reset();
+        const THREADS: usize = 8;
+        const PER_THREAD: u64 = 1_000;
+        std::thread::scope(|s| {
+            for _ in 0..THREADS {
+                s.spawn(|| {
+                    for _ in 0..PER_THREAD {
+                        LogEvent::DialFailure.record();
+                    }
+                });
+            }
+        });
+        let line = drain_and_format().expect("expected a summary");
+        assert!(line.contains(&format!("dial_failures={}", THREADS as u64 * PER_THREAD)));
+    }
+}