[Traffic Control] Improved error metrics and benchmarking

crates/sui-core/src/authority_server.rs

@@ -198,6 +198,7 @@ pub struct ValidatorServiceMetrics {
     forwarded_header_parse_error: IntCounter,
     forwarded_header_invalid: IntCounter,
     forwarded_header_not_included: IntCounter,
+    client_id_source_config_mismatch: IntCounter,
 }
 
 impl ValidatorServiceMetrics {
@@ -329,6 +330,12 @@ impl ValidatorServiceMetrics {
                 registry,
             )
             .unwrap(),
+            client_id_source_config_mismatch: register_int_counter_with_registry!(
+                "validator_service_client_id_source_config_mismatch",
+                "Number of times detected that client id source config doesn't agree with x-forwarded-for header",
+                registry,
+            )
+            .unwrap(),
         }
     }
 
@@ -1225,6 +1232,19 @@ impl ValidatorService {
                                 return None;
                             }
                             let contents_len = header_contents.len();
+                            if contents_len < *num_hops {
+                                error!(
+                                    "x-forwarded-for header value of {:?} contains {} values, but {} hops were specified. \
+                                    Expected at least {} values. Please correctly set the `x-forwarded-for` value under \
+                                    `client-id-source` in the node config.",
+                                    header_contents,
+                                    contents_len,
+                                    num_hops,
+                                    contents_len,
+                                );
+                                self.metrics.client_id_source_config_mismatch.inc();
+                                return None;
+                            }
                             let Some(client_ip) = header_contents.get(contents_len - num_hops)
                             else {
                                 error!(
@@ -1296,7 +1316,11 @@ impl ValidatorService {
             traffic_controller.tally(TrafficTally {
                 direct: client,
                 through_fullnode: None,
-                error_weight: error.map(normalize).unwrap_or(Weight::zero()),
+                error_info: error.map(|e| {
+                    let error_type = String::from(e.clone().as_ref());
+                    let error_weight = normalize(e);
+                    (error_weight, error_type)
+                }),
                 spam_weight,
                 timestamp: SystemTime::now(),
             })
@@ -1320,8 +1344,10 @@ fn make_tonic_request_for_testing<T>(message: T) -> tonic::Request<T> {
 // TODO: refine error matching here
 fn normalize(err: SuiError) -> Weight {
     match err {
-        SuiError::UserInputError { .. }
-        | SuiError::InvalidSignature { .. }
+        SuiError::UserInputError {
+            error: UserInputError::IncorrectUserSignature { .. },
+        } => Weight::one(),
+        SuiError::InvalidSignature { .. }
         | SuiError::SignerSignatureAbsent { .. }
         | SuiError::SignerSignatureNumberMismatch { .. }
         | SuiError::IncorrectSigner { .. }

crates/sui-core/src/traffic_controller/metrics.rs

@@ -2,8 +2,8 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use prometheus::{
-    register_int_counter_with_registry, register_int_gauge_with_registry, IntCounter, IntGauge,
-    Registry,
+    register_int_counter_vec_with_registry, register_int_counter_with_registry,
+    register_int_gauge_with_registry, IntCounter, IntCounterVec, IntGauge, Registry,
 };
 
 #[derive(Clone)]
@@ -18,6 +18,7 @@ pub struct TrafficControllerMetrics {
     pub num_dry_run_blocked_requests: IntCounter,
     pub tally_handled: IntCounter,
     pub error_tally_handled: IntCounter,
+    pub tally_error_types: IntCounterVec,
     pub deadmans_switch_enabled: IntGauge,
     pub highest_direct_spam_rate: IntGauge,
     pub highest_proxied_spam_rate: IntGauge,
@@ -90,6 +91,13 @@ impl TrafficControllerMetrics {
                 registry
             )
             .unwrap(),
+            tally_error_types: register_int_counter_vec_with_registry!(
+                "traffic_control_tally_error_types",
+                "Number of tally errors, grouped by error type",
+                &["error_type"],
+                registry
+            )
+            .unwrap(),
             deadmans_switch_enabled: register_int_gauge_with_registry!(
                 "deadmans_switch_enabled",
                 "If 1, the deadman's switch is enabled and all traffic control

crates/sui-core/src/traffic_controller/mod.rs

@@ -378,13 +378,13 @@ async fn run_tally_loop(
                     metrics
                         .highest_direct_spam_rate
                         .set(highest_direct_rate.0 as i64);
-                    trace!("Recent highest direct spam rate: {:?}", highest_direct_rate);
+                    debug!("Recent highest direct spam rate: {:?}", highest_direct_rate);
                 }
                 if let Some(highest_proxied_rate) = spam_policy.highest_proxied_rate() {
                     metrics
                         .highest_proxied_spam_rate
                         .set(highest_proxied_rate.0 as i64);
-                    trace!(
+                    debug!(
                         "Recent highest proxied spam rate: {:?}",
                         highest_proxied_rate
                     );
@@ -395,7 +395,7 @@ async fn run_tally_loop(
                     metrics
                         .highest_direct_error_rate
                         .set(highest_direct_rate.0 as i64);
-                    trace!(
+                    debug!(
                         "Recent highest direct error rate: {:?}",
                         highest_direct_rate
                     );
@@ -404,7 +404,7 @@ async fn run_tally_loop(
                     metrics
                         .highest_proxied_error_rate
                         .set(highest_proxied_rate.0 as i64);
-                    trace!(
+                    debug!(
                         "Recent highest proxied error rate: {:?}",
                         highest_proxied_rate
                     );
@@ -425,10 +425,22 @@ async fn handle_error_tally(
     metrics: Arc<TrafficControllerMetrics>,
     mem_drainfile_present: bool,
 ) -> Result<(), reqwest::Error> {
-    if !tally.error_weight.is_sampled() {
+    let Some((error_weight, error_type)) = tally.clone().error_info else {
+        return Ok(());
+    };
+    if !error_weight.is_sampled() {
         return Ok(());
     }
-    let resp = policy.handle_tally(tally.clone());
+    trace!(
+        "Handling error_type {:?} from client {:?}",
+        error_type,
+        tally.direct,
+    );
+    metrics
+        .tally_error_types
+        .with_label_values(&[error_type.as_str()])
+        .inc();
+    let resp = policy.handle_tally(tally);
     metrics.error_tally_handled.inc();
     if let Some(fw_config) = fw_config {
         if fw_config.delegate_error_blocking && !mem_drainfile_present {
@@ -509,6 +521,7 @@ async fn handle_policy_response(
         {
             // Only increment the metric if the client was not already blocked
             debug!("Blocking client: {:?}", client);
+            metrics.requests_blocked_at_protocol.inc();
             metrics.connection_ip_blocklist_len.inc();
         }
     }
@@ -523,6 +536,7 @@ async fn handle_policy_response(
         {
             // Only increment the metric if the client was not already blocked
             debug!("Blocking proxied client: {:?}", client);
+            metrics.requests_blocked_at_protocol.inc();
             metrics.proxy_ip_blocklist_len.inc();
         }
     }
@@ -745,7 +759,7 @@ impl TrafficSim {
                     // TODO add proxy IP for testing
                     None,
                     // TODO add weight adjustments
-                    Weight::one(),
+                    None,
                     Weight::one(),
                 ));
             } else {

crates/sui-core/src/traffic_controller/policies.rs

@@ -14,7 +14,7 @@ use std::hash::Hash;
 use std::time::Duration;
 use std::time::{Instant, SystemTime};
 use sui_types::traffic_control::{FreqThresholdConfig, PolicyConfig, PolicyType, Weight};
-use tracing::info;
+use tracing::{info, trace};
 
 const HIGHEST_RATES_CAPACITY: usize = 20;
 
@@ -222,7 +222,7 @@ impl TrafficSketch {
 pub struct TrafficTally {
     pub direct: Option<IpAddr>,
     pub through_fullnode: Option<IpAddr>,
-    pub error_weight: Weight,
+    pub error_info: Option<(Weight, String)>,
     pub spam_weight: Weight,
     pub timestamp: SystemTime,
 }
@@ -231,13 +231,13 @@ impl TrafficTally {
     pub fn new(
         direct: Option<IpAddr>,
         through_fullnode: Option<IpAddr>,
-        error_weight: Weight,
+        error_info: Option<(Weight, String)>,
         spam_weight: Weight,
     ) -> Self {
         Self {
             direct,
             through_fullnode,
-            error_weight,
+            error_info,
             spam_weight,
             timestamp: SystemTime::now(),
         }
@@ -360,7 +360,14 @@ impl FreqThresholdPolicy {
         let block_client = if let Some(source) = tally.direct {
             let key = SketchKey(source, ClientType::Direct);
             self.sketch.increment_count(&key);
-            if self.sketch.get_request_rate(&key) >= self.client_threshold as f64 {
+            let req_rate = self.sketch.get_request_rate(&key);
+            trace!(
+                "FreqThresholdPolicy handling tally -- req_rate: {:?}, client_threshold: {:?}, client: {:?}",
+                req_rate,
+                self.client_threshold,
+                source,
+            );
+            if req_rate >= self.client_threshold as f64 {
                 Some(source)
             } else {
                 None
@@ -515,21 +522,21 @@ mod tests {
         let alice = TrafficTally {
             direct: Some(IpAddr::V4(Ipv4Addr::new(8, 7, 6, 5))),
             through_fullnode: Some(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4))),
-            error_weight: Weight::zero(),
+            error_info: None,
             spam_weight: Weight::one(),
             timestamp: SystemTime::now(),
         };
         let bob = TrafficTally {
             direct: Some(IpAddr::V4(Ipv4Addr::new(8, 7, 6, 5))),
             through_fullnode: Some(IpAddr::V4(Ipv4Addr::new(4, 3, 2, 1))),
-            error_weight: Weight::zero(),
+            error_info: None,
             spam_weight: Weight::one(),
             timestamp: SystemTime::now(),
         };
         let charlie = TrafficTally {
             direct: Some(IpAddr::V4(Ipv4Addr::new(8, 7, 6, 5))),
             through_fullnode: Some(IpAddr::V4(Ipv4Addr::new(5, 6, 7, 8))),
-            error_weight: Weight::zero(),
+            error_info: None,
             spam_weight: Weight::one(),
             timestamp: SystemTime::now(),
         };

crates/sui-e2e-tests/tests/traffic_control_tests.rs

@@ -26,7 +26,9 @@ use sui_network::default_mysten_network_config;
 use sui_swarm_config::network_config_builder::ConfigBuilder;
 use sui_test_transaction_builder::batch_make_transfer_transactions;
 use sui_types::{
+    crypto::Ed25519SuiSignature,
     quorum_driver_types::ExecuteTransactionRequestType,
+    signature::GenericSignature,
     traffic_control::{
         FreqThresholdConfig, PolicyConfig, PolicyType, RemoteFirewallConfig, Weight,
     },
@@ -225,7 +227,7 @@ async fn test_validator_traffic_control_error_blocked() -> Result<(), anyhow::Er
         .with_policy_config(Some(policy_config))
         .build();
     let committee = network_config.committee_with_network();
-    let _test_cluster = TestClusterBuilder::new()
+    let test_cluster = TestClusterBuilder::new()
         .set_network_config(network_config)
         .build()
         .await;
@@ -235,12 +237,13 @@ async fn test_validator_traffic_control_error_blocked() -> Result<(), anyhow::Er
     );
     let (_, auth_client) = local_clients.first_key_value().unwrap();
 
-    // transaction signed using user wallet from a different chain/genesis,
-    // therefore we should fail with UserInputError
-    let other_cluster = TestClusterBuilder::new().build().await;
-
-    let mut txns = batch_make_transfer_transactions(&other_cluster.wallet, n as usize).await;
-    let tx = txns.swap_remove(0);
+    let mut txns = batch_make_transfer_transactions(&test_cluster.wallet, n as usize).await;
+    let mut tx = txns.swap_remove(0);
+    let signatures = tx.tx_signatures_mut_for_testing();
+    signatures.pop();
+    signatures.push(GenericSignature::Signature(
+        sui_types::crypto::Signature::Ed25519SuiSignature(Ed25519SuiSignature::default()),
+    ));
 
     // it should take no more than 4 requests to be added to the blocklist
     for _ in 0..n {
@@ -251,7 +254,7 @@ async fn test_validator_traffic_control_error_blocked() -> Result<(), anyhow::Er
             }
         }
     }
-    panic!("Expected spam policy to trigger within {n} requests");
+    panic!("Expected error policy to trigger within {n} requests");
 }
 
 #[tokio::test]
@@ -406,7 +409,7 @@ async fn test_validator_traffic_control_error_delegated() -> Result<(), anyhow::
         .with_firewall_config(Some(firewall_config))
         .build();
     let committee = network_config.committee_with_network();
-    let _test_cluster = TestClusterBuilder::new()
+    let test_cluster = TestClusterBuilder::new()
         .set_network_config(network_config)
         .build()
         .await;
@@ -416,12 +419,13 @@ async fn test_validator_traffic_control_error_delegated() -> Result<(), anyhow::
     );
     let (_, auth_client) = local_clients.first_key_value().unwrap();
 
-    // transaction signed using user wallet from a different chain/genesis,
-    // therefore we should fail with UserInputError
-    let other_cluster = TestClusterBuilder::new().build().await;
-
-    let mut txns = batch_make_transfer_transactions(&other_cluster.wallet, n as usize).await;
-    let tx = txns.swap_remove(0);
+    let mut txns = batch_make_transfer_transactions(&test_cluster.wallet, n as usize).await;
+    let mut tx = txns.swap_remove(0);
+    let signatures = tx.tx_signatures_mut_for_testing();
+    signatures.pop();
+    signatures.push(GenericSignature::Signature(
+        sui_types::crypto::Signature::Ed25519SuiSignature(Ed25519SuiSignature::default()),
+    ));
 
     // start test firewall server
     let mut server = NodeFwTestServer::new();

crates/sui-json-rpc/src/axum_router.rs

@@ -252,7 +252,11 @@ fn handle_traffic_resp(
     traffic_controller.tally(TrafficTally {
         direct: client,
         through_fullnode: None,
-        error_weight: error.map(normalize).unwrap_or(Weight::zero()),
+        error_info: error.map(|e| {
+            let error_type = e.to_string();
+            let error_weight = normalize(e);
+            (error_weight, error_type)
+        }),
         // For now, count everything as spam with equal weight
         // on the rpc node side, including gas-charging endpoints
         // such as `sui_executeTransactionBlock`, as this can enable