[Consensus] add tracing and basic metrics to tonic

consensus/config/src/parameters.rs

@@ -182,11 +182,13 @@ pub struct TonicParameters {
     #[serde(default = "TonicParameters::default_connection_buffer_size")]
     pub connection_buffer_size: usize,
 
-    /// Message size limits for both requests and responses.
+    /// Hard message size limit for both requests and responses.
+    /// This value is higher than strictly necessary, to allow overheads.
+    /// Message size targets and soft limits are computed based on this value.
     ///
-    /// If unspecified, this will default to 8MiB.
+    /// If unspecified, this will default to 32MiB.
     #[serde(default = "TonicParameters::default_message_size_limit")]
-    pub message_size_limit: usize,
+    message_size_limit: usize,
 }
 
 impl TonicParameters {
@@ -199,7 +201,15 @@ impl TonicParameters {
     }
 
     fn default_message_size_limit() -> usize {
-        8 << 20
+        32 << 20
+    }
+
+    pub fn excessive_message_size(&self) -> usize {
+        self.message_size_limit * 3 / 4
+    }
+
+    pub fn max_message_size(&self) -> usize {
+        self.message_size_limit
     }
 }
 

consensus/config/tests/snapshots/parameters_test__parameters.snap

@@ -24,4 +24,4 @@ tonic:
     secs: 5
     nanos: 0
   connection_buffer_size: 33554432
-  message_size_limit: 8388608
+  message_size_limit: 33554432

consensus/core/Cargo.toml

@@ -21,6 +21,7 @@ dashmap.workspace = true
 enum_dispatch.workspace = true
 fastcrypto.workspace = true
 futures.workspace = true
+http.workspace = true
 hyper.workspace = true
 hyper-rustls.workspace = true
 itertools.workspace = true

consensus/core/src/metrics.rs

@@ -10,7 +10,7 @@ use prometheus::{
     HistogramVec, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, Registry,
 };
 
-use crate::network::metrics::{NetworkRouteMetrics, QuinnConnectionMetrics};
+use crate::network::metrics::{NetworkMetrics, QuinnConnectionMetrics};
 
 // starts from 1μs, 50μs, 100μs...
 const FINE_GRAINED_LATENCY_SEC_BUCKETS: &[f64] = &[
@@ -486,26 +486,3 @@ impl ChannelMetrics {
         }
     }
 }
-
-// Fields for network-agnostic metrics can be added here
-pub(crate) struct NetworkMetrics {
-    pub(crate) network_type: IntGaugeVec,
-    pub(crate) inbound: NetworkRouteMetrics,
-    pub(crate) outbound: NetworkRouteMetrics,
-}
-
-impl NetworkMetrics {
-    pub(crate) fn new(registry: &Registry) -> Self {
-        Self {
-            network_type: register_int_gauge_vec_with_registry!(
-                "network_type",
-                "Type of the network used: anemo or tonic",
-                &["type"],
-                registry
-            )
-            .unwrap(),
-            inbound: NetworkRouteMetrics::new("inbound", registry),
-            outbound: NetworkRouteMetrics::new("outbound", registry),
-        }
-    }
-}

consensus/core/src/network/anemo_network.rs

@@ -22,7 +22,6 @@ use async_trait::async_trait;
 use bytes::Bytes;
 use cfg_if::cfg_if;
 use consensus_config::{AuthorityIndex, NetworkKeyPair};
-use prometheus::HistogramTimer;
 use serde::{Deserialize, Serialize};
 use tokio::sync::broadcast::error::RecvError;
 use tracing::{debug, error, warn};
@@ -34,7 +33,7 @@ use super::{
     },
     connection_monitor::{AnemoConnectionMonitor, ConnectionMonitorHandle},
     epoch_filter::{AllowedEpoch, EPOCH_HEADER_KEY},
-    metrics::NetworkRouteMetrics,
+    metrics_layer::{MetricsCallbackMaker, MetricsResponseCallback, SizedRequest, SizedResponse},
     BlockStream, NetworkClient, NetworkManager, NetworkService,
 };
 use crate::{
@@ -403,10 +402,8 @@ impl<S: NetworkService> NetworkManager<S> for AnemoManager {
             }
         );
         let epoch_string: String = self.context.committee.epoch().to_string();
-        let inbound_network_metrics =
-            Arc::new(self.context.metrics.network_metrics.inbound.clone());
-        let outbound_network_metrics =
-            Arc::new(self.context.metrics.network_metrics.outbound.clone());
+        let inbound_network_metrics = self.context.metrics.network_metrics.inbound.clone();
+        let outbound_network_metrics = self.context.metrics.network_metrics.outbound.clone();
         let quinn_connection_metrics = self.context.metrics.quinn_connection_metrics.clone();
         let all_peer_ids = self
             .context
@@ -430,7 +427,7 @@ impl<S: NetworkService> NetworkManager<S> for AnemoManager {
                     .make_span_with(DefaultMakeSpan::new().level(tracing::Level::INFO))
                     .on_failure(DefaultOnFailure::new().level(tracing::Level::WARN)),
             )
-            .layer(CallbackLayer::new(MetricsMakeCallbackHandler::new(
+            .layer(CallbackLayer::new(MetricsCallbackMaker::new(
                 inbound_network_metrics,
                 self.context.parameters.anemo.excessive_message_size,
             )))
@@ -446,7 +443,7 @@ impl<S: NetworkService> NetworkManager<S> for AnemoManager {
                     .make_span_with(DefaultMakeSpan::new().level(tracing::Level::INFO))
                     .on_failure(DefaultOnFailure::new().level(tracing::Level::WARN)),
             )
-            .layer(CallbackLayer::new(MetricsMakeCallbackHandler::new(
+            .layer(CallbackLayer::new(MetricsCallbackMaker::new(
                 outbound_network_metrics,
                 self.context.parameters.anemo.excessive_message_size,
             )))
@@ -576,6 +573,50 @@ impl<S: NetworkService> NetworkManager<S> for AnemoManager {
     }
 }
 
+// Adapt MetricsCallbackMaker and MetricsResponseCallback to anemo.
+
+impl SizedRequest for anemo::Request<Bytes> {
+    fn size(&self) -> usize {
+        self.body().len()
+    }
+
+    fn route(&self) -> String {
+        self.route().to_string()
+    }
+}
+
+impl SizedResponse for anemo::Response<Bytes> {
+    fn size(&self) -> usize {
+        self.body().len()
+    }
+
+    fn error_type(&self) -> Option<String> {
+        if self.status().is_success() {
+            None
+        } else {
+            Some(self.status().to_string())
+        }
+    }
+}
+
+impl MakeCallbackHandler for MetricsCallbackMaker {
+    type Handler = MetricsResponseCallback;
+
+    fn make_handler(&self, request: &anemo::Request<bytes::Bytes>) -> Self::Handler {
+        self.handle_request(request)
+    }
+}
+
+impl ResponseHandler for MetricsResponseCallback {
+    fn on_response(self, response: &anemo::Response<bytes::Bytes>) {
+        self.on_response(response)
+    }
+
+    fn on_error<E>(self, err: &E) {
+        self.on_error(err)
+    }
+}
+
 /// Network message types.
 #[derive(Clone, Serialize, Deserialize)]
 pub(crate) struct SendBlockRequest {
@@ -611,115 +652,3 @@ pub(crate) struct FetchCommitsResponse {
     // Serialized SignedBlock that certify the last commit from above.
     certifier_blocks: Vec<Bytes>,
 }
-
-#[derive(Clone)]
-pub(crate) struct MetricsMakeCallbackHandler {
-    metrics: Arc<NetworkRouteMetrics>,
-    /// Size in bytes above which a request or response message is considered excessively large
-    excessive_message_size: usize,
-}
-
-impl MetricsMakeCallbackHandler {
-    pub fn new(metrics: Arc<NetworkRouteMetrics>, excessive_message_size: usize) -> Self {
-        Self {
-            metrics,
-            excessive_message_size,
-        }
-    }
-}
-
-impl MakeCallbackHandler for MetricsMakeCallbackHandler {
-    type Handler = MetricsResponseHandler;
-
-    fn make_handler(&self, request: &anemo::Request<bytes::Bytes>) -> Self::Handler {
-        let route = request.route().to_owned();
-
-        self.metrics.requests.with_label_values(&[&route]).inc();
-        self.metrics
-            .inflight_requests
-            .with_label_values(&[&route])
-            .inc();
-        let body_len = request.body().len();
-        self.metrics
-            .request_size
-            .with_label_values(&[&route])
-            .observe(body_len as f64);
-        if body_len > self.excessive_message_size {
-            warn!(
-                "Saw excessively large request with size {body_len} for {route} with peer {:?}",
-                request.peer_id()
-            );
-            self.metrics
-                .excessive_size_requests
-                .with_label_values(&[&route])
-                .inc();
-        }
-
-        let timer = self
-            .metrics
-            .request_latency
-            .with_label_values(&[&route])
-            .start_timer();
-
-        MetricsResponseHandler {
-            metrics: self.metrics.clone(),
-            timer,
-            route,
-            excessive_message_size: self.excessive_message_size,
-        }
-    }
-}
-
-pub(crate) struct MetricsResponseHandler {
-    metrics: Arc<NetworkRouteMetrics>,
-    // The timer is held on to and "observed" once dropped
-    #[allow(unused)]
-    timer: HistogramTimer,
-    route: String,
-    excessive_message_size: usize,
-}
-
-impl ResponseHandler for MetricsResponseHandler {
-    fn on_response(self, response: &anemo::Response<bytes::Bytes>) {
-        let body_len = response.body().len();
-        self.metrics
-            .response_size
-            .with_label_values(&[&self.route])
-            .observe(body_len as f64);
-        if body_len > self.excessive_message_size {
-            warn!(
-                "Saw excessively large response with size {body_len} for {} with peer {:?}",
-                self.route,
-                response.peer_id()
-            );
-            self.metrics
-                .excessive_size_responses
-                .with_label_values(&[&self.route])
-                .inc();
-        }
-
-        if !response.status().is_success() {
-            let status = response.status().to_u16().to_string();
-            self.metrics
-                .errors
-                .with_label_values(&[&self.route, &status])
-                .inc();
-        }
-    }
-
-    fn on_error<E>(self, _error: &E) {
-        self.metrics
-            .errors
-            .with_label_values(&[&self.route, "unknown"])
-            .inc();
-    }
-}
-
-impl Drop for MetricsResponseHandler {
-    fn drop(&mut self) {
-        self.metrics
-            .inflight_requests
-            .with_label_values(&[&self.route])
-            .dec();
-    }
-}

consensus/core/src/network/metrics.rs

@@ -1,12 +1,37 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
+use std::sync::Arc;
+
 use prometheus::{
     register_histogram_vec_with_registry, register_int_counter_vec_with_registry,
     register_int_gauge_vec_with_registry, register_int_gauge_with_registry, HistogramVec,
     IntCounterVec, IntGauge, IntGaugeVec, Registry,
 };
 
+// Fields for network-agnostic metrics can be added here
+pub(crate) struct NetworkMetrics {
+    pub(crate) network_type: IntGaugeVec,
+    pub(crate) inbound: Arc<NetworkRouteMetrics>,
+    pub(crate) outbound: Arc<NetworkRouteMetrics>,
+}
+
+impl NetworkMetrics {
+    pub(crate) fn new(registry: &Registry) -> Self {
+        Self {
+            network_type: register_int_gauge_vec_with_registry!(
+                "network_type",
+                "Type of the network used: anemo or tonic",
+                &["type"],
+                registry
+            )
+            .unwrap(),
+            inbound: Arc::new(NetworkRouteMetrics::new("inbound", registry)),
+            outbound: Arc::new(NetworkRouteMetrics::new("outbound", registry)),
+        }
+    }
+}
+
 #[derive(Clone)]
 pub(crate) struct QuinnConnectionMetrics {
     /// The connection status of known peers. 0 if not connected, 1 if connected.