Track dynode retries separately

Author: gemcoder21

core/apps/dynode/src/metrics/mod.rs

@@ -13,6 +13,8 @@ pub struct Metrics {
     proxy_requests: Family<ProxyRequestLabels, Counter>,
     proxy_requests_by_method: Family<ProxyRequestByMethodLabels, Counter>,
     proxy_response_latency: Family<ResponseLabels, Histogram>,
+    proxy_upstream_response_latency: Family<UpstreamResponseLabels, Histogram>,
+    proxy_retries: Family<RetryLabels, Counter>,
     node_host_current: Family<HostCurrentStateLabels, Gauge>,
     cache_hits: Family<CacheLabels, Counter>,
     cache_misses: Family<CacheLabels, Counter>,
@@ -41,12 +43,25 @@ pub struct HostCurrentStateLabels {
 
 #[derive(Clone, Hash, PartialEq, Eq, Debug, EncodeLabelSet)]
 struct ResponseLabels {
+    chain: String,
+    status: u16,
+}
+
+#[derive(Clone, Hash, PartialEq, Eq, Debug, EncodeLabelSet)]
+struct UpstreamResponseLabels {
     chain: String,
     host: String,
     method: String,
     status: u16,
 }
 
+#[derive(Clone, Hash, PartialEq, Eq, Debug, EncodeLabelSet)]
+struct RetryLabels {
+    chain: String,
+    host: String,
+    reason: String,
+}
+
 #[derive(Clone, Hash, PartialEq, Eq, Debug, EncodeLabelSet)]
 pub struct CacheLabels {
     chain: String,
@@ -71,6 +86,8 @@ impl Metrics {
         let proxy_requests = Family::<ProxyRequestLabels, Counter>::default();
         let proxy_requests_by_method = Family::<ProxyRequestByMethodLabels, Counter>::default();
         let proxy_response_latency = Family::<ResponseLabels, Histogram>::new_with_constructor(|| Histogram::new(exponential_buckets(50.0, 2.0, 6)));
+        let proxy_upstream_response_latency = Family::<UpstreamResponseLabels, Histogram>::new_with_constructor(|| Histogram::new(exponential_buckets(50.0, 2.0, 6)));
+        let proxy_retries = Family::<RetryLabels, Counter>::default();
         let node_host_current = Family::<HostCurrentStateLabels, Gauge>::default();
         let cache_hits = Family::<CacheLabels, Counter>::default();
         let cache_misses = Family::<CacheLabels, Counter>::default();
@@ -87,11 +104,13 @@ impl Metrics {
             "Proxy requests by host and method (HTTP path or RPC method)",
             proxy_requests_by_method.clone(),
         );
+        registry.register("proxy_response_latency", "Proxy responses by chain and status", proxy_response_latency.clone());
         registry.register(
-            "proxy_response_latency",
-            "Proxy responses by host, path, method, and status",
-            proxy_response_latency.clone(),
+            "proxy_upstream_response_latency",
+            "Upstream proxy responses by host, path, method, and status",
+            proxy_upstream_response_latency.clone(),
         );
+        registry.register("proxy_retries", "Proxy retries by chain, upstream host, and reason", proxy_retries.clone());
         registry.register("node_host_current", "Node current host url", node_host_current.clone());
         registry.register("cache_hits", "Cache hits by host and path", cache_hits.clone());
         registry.register("cache_misses", "Cache misses by host and path", cache_misses.clone());
@@ -105,6 +124,8 @@ impl Metrics {
             proxy_requests,
             proxy_requests_by_method,
             proxy_response_latency,
+            proxy_upstream_response_latency,
+            proxy_retries,
             node_host_current,
             cache_hits,
             cache_misses,
@@ -137,10 +158,10 @@ impl Metrics {
         }
     }
 
-    pub fn add_proxy_response(&self, chain: &str, method: &str, host: &str, status: u16, latency: u128) {
+    pub fn add_proxy_upstream_response(&self, chain: &str, method: &str, host: &str, status: u16, latency: u128) {
         let method = self.truncate_method(method);
-        self.proxy_response_latency
-            .get_or_create(&ResponseLabels {
+        self.proxy_upstream_response_latency
+            .get_or_create(&UpstreamResponseLabels {
                 chain: chain.to_string(),
                 host: host.to_string(),
                 method,
@@ -149,6 +170,22 @@ impl Metrics {
             .observe(latency as f64);
     }
 
+    pub fn add_proxy_response(&self, chain: &str, status: u16, latency: u128) {
+        self.proxy_response_latency
+            .get_or_create(&ResponseLabels { chain: chain.to_string(), status })
+            .observe(latency as f64);
+    }
+
+    pub fn add_proxy_retry(&self, chain: &str, host: &str, reason: &str) {
+        self.proxy_retries
+            .get_or_create(&RetryLabels {
+                chain: chain.to_string(),
+                host: host.to_string(),
+                reason: reason.to_string(),
+            })
+            .inc();
+    }
+
     pub fn set_node_host_current(&self, chain: &str, host: &str) {
         self.node_host_current
             .get_or_create(&HostCurrentStateLabels {
@@ -266,4 +303,20 @@ mod tests {
         assert_eq!(m.truncate_method("/api/v1/blocks/by_height/12345"), "/api/v1/blocks/by_height/:number");
         assert_eq!(m.truncate_method("/v1/verylongsegmentthatisgreaterthan20characters"), "/v1/:value");
     }
+
+    #[test]
+    fn records_response_once_and_retries_separately() {
+        let metrics = create_test_metrics();
+
+        metrics.add_proxy_response("tron", 200, 123);
+        metrics.add_proxy_retry("tron", "api.trongrid.io", "status=429");
+
+        let encoded = metrics.get_metrics();
+        assert!(encoded.contains("test_proxy_response_latency_count"));
+        assert!(encoded.contains("chain=\"tron\""));
+        assert!(encoded.contains("status=\"200\""));
+        assert!(encoded.contains("test_proxy_retries_total"));
+        assert!(encoded.contains("host=\"api.trongrid.io\""));
+        assert!(encoded.contains("reason=\"status=429\""));
+    }
 }

core/apps/dynode/src/monitoring/service.rs

@@ -101,6 +101,16 @@ impl NodeService {
     }
 
     pub async fn handle_request(&self, request: ProxyRequest) -> Result<ProxyResponse, Box<dyn Error + Send + Sync>> {
+        let request_for_metrics = request.clone();
+        let result = self.handle_request_inner(request).await;
+        if let Ok(response) = &result {
+            self.metrics
+                .add_proxy_response(request_for_metrics.chain.as_ref(), response.status, request_for_metrics.elapsed().as_millis());
+        }
+        result
+    }
+
+    async fn handle_request_inner(&self, request: ProxyRequest) -> Result<ProxyResponse, Box<dyn Error + Send + Sync>> {
         Self::log_incoming_request(&request);
 
         let chain_config = self.get_chain_config(&request)?;
@@ -144,7 +154,11 @@ impl NodeService {
                     if !retry_enabled {
                         return self.log_and_create_error_response(&request, Some(remote_host.as_str()), &format!("Upstream status code: {}", response.status), upstream_data);
                     }
-                    last_error = Some(format!("status={}", response.status));
+                    let retry_reason = format!("status={}", response.status);
+                    if index + 1 < max_attempts {
+                        self.metrics.add_proxy_retry(request.chain.as_ref(), remote_host.as_str(), &retry_reason);
+                    }
+                    last_error = Some(retry_reason);
                     last_error_data = upstream_data;
                 }
                 Err(e) => {
@@ -163,6 +177,9 @@ impl NodeService {
                         error = UPSTREAM_REQUEST_FAILED,
                         latency = latency,
                     );
+                    if index + 1 < max_attempts {
+                        self.metrics.add_proxy_retry(request.chain.as_ref(), remote_host.as_str(), UPSTREAM_REQUEST_FAILED);
+                    }
                     last_error = Some(UPSTREAM_REQUEST_FAILED.to_string());
                 }
             }

core/apps/dynode/src/proxy/jsonrpc.rs

@@ -68,7 +68,7 @@ impl JsonRpcHandler {
                     id: Some(call.id),
                 });
 
-                metrics.add_proxy_response(
+                metrics.add_proxy_upstream_response(
                     request.chain.as_ref(),
                     &call.method,
                     url.url.host_str().unwrap_or_default(),
@@ -87,7 +87,7 @@ impl JsonRpcHandler {
 
         let (response, response_status, response_body) = Self::fetch_single_response(call, request, cache, url, client, forward_headers).await?;
 
-        metrics.add_proxy_response(
+        metrics.add_proxy_upstream_response(
             request.chain.as_ref(),
             &call.method,
             url.url.host_str().unwrap_or_default(),
@@ -150,7 +150,7 @@ impl JsonRpcHandler {
         let (responses, response_status) = Self::fetch_batch_responses(calls, url, client, &request.method, forward_headers).await?;
 
         for call in calls {
-            metrics.add_proxy_response(
+            metrics.add_proxy_upstream_response(
                 request.chain.as_ref(),
                 &call.method,
                 url.url.host_str().unwrap_or_default(),

core/apps/dynode/src/proxy/service.rs

@@ -99,7 +99,7 @@ impl ProxyRequestService {
 
     fn add_proxy_response_metrics(metrics: &Metrics, request: &ProxyRequest, methods_for_metrics: &[String], host: &str, status: u16) {
         for method_name in methods_for_metrics {
-            metrics.add_proxy_response(request.chain.as_ref(), method_name, host, status, request.elapsed().as_millis());
+            metrics.add_proxy_upstream_response(request.chain.as_ref(), method_name, host, status, request.elapsed().as_millis());
         }
     }