feat(indexer,gateway): add and unify service metrics (#517)

* indexer: add core operational metrics

- add chain head/processed gauges and websocket reconnect counter
- add commit and tree-sync latency/size histograms
- add HTTP latency middleware with route/status-class tags
- register metrics on startup and wire instrumentation through indexer flows

* indexer: remove ws reconnect metric

Drop indexer.ws.reconnects and related reconnect counter updates.
Keep chain/head, processed, tree sync, commit, and HTTP latency metrics.

* gateway: unify metrics API and HTTP latency naming

Refactor gateway metrics emission through helper functions in metrics.rs,
register metric descriptions at startup, and align HTTP latency naming/tags
with the indexer approach (gateway.http.latency_ms, route, status_class).

* metrics: remove service-name prefixes

- drop gateway./indexer. prefixes from metric keys
- keep namespaced paths without service prefix (e.g. http.latency_ms)
- clean up indexer middleware by removing no-op path normalizer

Author: Dzejkop

Cargo.lock

@@ -8,14 +8,7 @@ use alloy::{
     providers::{DynProvider, Provider},
 };
 
-use crate::{
-    config::BatchPolicyConfig,
-    metrics::{
-        METRICS_BATCH_POLICY_COST_SCORE, METRICS_BATCH_POLICY_DEFER,
-        METRICS_BATCH_POLICY_FORCE_SEND, METRICS_BATCH_POLICY_TARGET_SIZE,
-        METRICS_BATCH_POLICY_URGENCY_SCORE,
-    },
-};
+use crate::{config::BatchPolicyConfig, metrics};
 
 /// Aggregated queued backlog pressure from Redis.
 #[derive(Debug, Clone, Copy, Default)]
@@ -272,24 +265,19 @@ impl BatchPolicyEngine {
 
 /// Emits policy metrics for a decision.
 pub fn record_policy_metrics(batch_type: &'static str, decision: &PolicyDecision) {
-    ::metrics::histogram!(METRICS_BATCH_POLICY_COST_SCORE, "type" => batch_type)
-        .record(decision.cost_score);
-    ::metrics::histogram!(METRICS_BATCH_POLICY_URGENCY_SCORE, "type" => batch_type)
-        .record(decision.urgency_score);
-    ::metrics::histogram!(METRICS_BATCH_POLICY_TARGET_SIZE, "type" => batch_type)
-        .record(decision.target_batch_size as f64);
+    metrics::record_policy_scores(
+        batch_type,
+        decision.cost_score,
+        decision.urgency_score,
+        decision.target_batch_size,
+    );
 
     if decision.force_send {
-        ::metrics::counter!(METRICS_BATCH_POLICY_FORCE_SEND, "type" => batch_type).increment(1);
+        metrics::increment_policy_force_send(batch_type);
     }
 
     if !decision.should_send {
-        ::metrics::counter!(
-            METRICS_BATCH_POLICY_DEFER,
-            "type" => batch_type,
-            "reason" => decision.reason.as_str()
-        )
-        .increment(1);
+        metrics::increment_policy_defer(batch_type, decision.reason.as_str());
     }
 }
 

services/gateway/src/batch_policy.rs

@@ -5,10 +5,7 @@ use crate::{
     batch_policy::BaseFeeCache,
     config::BatchPolicyConfig,
     error::parse_contract_error,
-    metrics::{
-        METRICS_BATCH_FAILURE, METRICS_BATCH_LATENCY_MS, METRICS_BATCH_SIZE,
-        METRICS_BATCH_SUBMITTED, METRICS_BATCH_SUCCESS,
-    },
+    metrics,
     policy_batcher::{PolicyBatchLoopRunner, TimedEnvelope},
     request_tracker::BacklogScope,
 };
@@ -92,8 +89,7 @@ impl CreateBatcherRunner {
         let batch_size = batch.len();
         let ids: Vec<String> = batch.iter().map(|env| env.id.clone()).collect();
 
-        ::metrics::counter!(METRICS_BATCH_SUBMITTED, "type" => "create").increment(1);
-        ::metrics::histogram!(METRICS_BATCH_SIZE, "type" => "create").record(batch_size as f64);
+        metrics::record_batch_submitted("create", batch_size);
 
         self.tracker
             .set_status_batch(&ids, GatewayRequestState::Batching)
@@ -122,9 +118,7 @@ impl CreateBatcherRunner {
         match call.send().await {
             Ok(builder) => {
                 let latency_ms = start.elapsed().as_millis() as f64;
-                ::metrics::histogram!(METRICS_BATCH_LATENCY_MS, "type" => "create")
-                    .record(latency_ms);
-                ::metrics::counter!(METRICS_BATCH_SUCCESS, "type" => "create").increment(1);
+                metrics::record_batch_result("create", true, latency_ms);
 
                 let hash = format!("0x{:x}", builder.tx_hash());
                 self.tracker
@@ -164,9 +158,7 @@ impl CreateBatcherRunner {
             }
             Err(err) => {
                 let latency_ms = start.elapsed().as_millis() as f64;
-                ::metrics::histogram!(METRICS_BATCH_LATENCY_MS, "type" => "create")
-                    .record(latency_ms);
-                ::metrics::counter!(METRICS_BATCH_FAILURE, "type" => "create").increment(1);
+                metrics::record_batch_result("create", false, latency_ms);
 
                 tracing::error!(error = %err, "create batch send failed");
                 let error_str = err.to_string();

services/gateway/src/create_batcher.rs

@@ -17,7 +17,7 @@ mod batcher;
 mod config;
 mod create_batcher;
 mod error;
-mod metrics;
+pub mod metrics;
 pub mod nonce;
 mod ops_batcher;
 pub mod orphan_sweeper;

services/gateway/src/lib.rs

@@ -7,6 +7,7 @@ async fn main() -> GatewayResult<()> {
     let env_path = Path::new(env!("CARGO_MANIFEST_DIR")).join(".env"); // load env vars in the root of this service
     let _ = dotenvy::from_path(&env_path);
     let _guard = telemetry_batteries::init();
+    world_id_gateway::metrics::describe_metrics();
 
     let _ = dotenvy::dotenv();
     tracing::info!("Starting world-id-gateway");

services/gateway/src/main.rs

@@ -1,20 +1,162 @@
-//! Metrics definitions for the world-id-gateway.
+//! Metrics definitions and helpers for the world-id-gateway.
 
-// Request metrics
-pub const METRICS_REQUESTS_LATENCY_MS: &str = "gateway.requests.latency_ms";
+// HTTP metrics
+pub const METRICS_HTTP_LATENCY_MS: &str = "http.latency_ms";
 
 // Root cache metrics
-pub const METRICS_ROOT_CACHE_HITS: &str = "gateway.root_cache.hits";
-pub const METRICS_ROOT_CACHE_MISSES: &str = "gateway.root_cache.misses";
+pub const METRICS_ROOT_CACHE_HITS: &str = "root_cache.hits";
+pub const METRICS_ROOT_CACHE_MISSES: &str = "root_cache.misses";
 
 // Batcher metrics
-pub const METRICS_BATCH_SUBMITTED: &str = "gateway.batch.submitted";
-pub const METRICS_BATCH_SIZE: &str = "gateway.batch.size";
-pub const METRICS_BATCH_LATENCY_MS: &str = "gateway.batch.latency_ms";
-pub const METRICS_BATCH_SUCCESS: &str = "gateway.batch.success";
-pub const METRICS_BATCH_FAILURE: &str = "gateway.batch.failure";
-pub const METRICS_BATCH_POLICY_COST_SCORE: &str = "gateway.batch.policy.cost_score";
-pub const METRICS_BATCH_POLICY_URGENCY_SCORE: &str = "gateway.batch.policy.urgency_score";
-pub const METRICS_BATCH_POLICY_DEFER: &str = "gateway.batch.policy.defer";
-pub const METRICS_BATCH_POLICY_FORCE_SEND: &str = "gateway.batch.policy.force_send";
-pub const METRICS_BATCH_POLICY_TARGET_SIZE: &str = "gateway.batch.policy.target_size";
+pub const METRICS_BATCH_SUBMITTED: &str = "batch.submitted";
+pub const METRICS_BATCH_SIZE: &str = "batch.size";
+pub const METRICS_BATCH_LATENCY_MS: &str = "batch.latency_ms";
+pub const METRICS_BATCH_SUCCESS: &str = "batch.success";
+pub const METRICS_BATCH_FAILURE: &str = "batch.failure";
+pub const METRICS_BATCH_POLICY_COST_SCORE: &str = "batch.policy.cost_score";
+pub const METRICS_BATCH_POLICY_URGENCY_SCORE: &str = "batch.policy.urgency_score";
+pub const METRICS_BATCH_POLICY_DEFER: &str = "batch.policy.defer";
+pub const METRICS_BATCH_POLICY_FORCE_SEND: &str = "batch.policy.force_send";
+pub const METRICS_BATCH_POLICY_TARGET_SIZE: &str = "batch.policy.target_size";
+
+pub fn describe_metrics() {
+    ::metrics::describe_histogram!(
+        METRICS_HTTP_LATENCY_MS,
+        ::metrics::Unit::Milliseconds,
+        "Gateway HTTP request latency in milliseconds."
+    );
+
+    ::metrics::describe_counter!(
+        METRICS_ROOT_CACHE_HITS,
+        ::metrics::Unit::Count,
+        "Number of root cache hits."
+    );
+    ::metrics::describe_counter!(
+        METRICS_ROOT_CACHE_MISSES,
+        ::metrics::Unit::Count,
+        "Number of root cache misses."
+    );
+
+    ::metrics::describe_counter!(
+        METRICS_BATCH_SUBMITTED,
+        ::metrics::Unit::Count,
+        "Number of submitted batches."
+    );
+    ::metrics::describe_histogram!(
+        METRICS_BATCH_SIZE,
+        ::metrics::Unit::Count,
+        "Number of requests per submitted batch."
+    );
+    ::metrics::describe_histogram!(
+        METRICS_BATCH_LATENCY_MS,
+        ::metrics::Unit::Milliseconds,
+        "Batch submission latency in milliseconds."
+    );
+    ::metrics::describe_counter!(
+        METRICS_BATCH_SUCCESS,
+        ::metrics::Unit::Count,
+        "Number of successfully submitted batches."
+    );
+    ::metrics::describe_counter!(
+        METRICS_BATCH_FAILURE,
+        ::metrics::Unit::Count,
+        "Number of failed batch submissions."
+    );
+
+    ::metrics::describe_histogram!(
+        METRICS_BATCH_POLICY_COST_SCORE,
+        ::metrics::Unit::Count,
+        "Batch policy cost score."
+    );
+    ::metrics::describe_histogram!(
+        METRICS_BATCH_POLICY_URGENCY_SCORE,
+        ::metrics::Unit::Count,
+        "Batch policy urgency score."
+    );
+    ::metrics::describe_histogram!(
+        METRICS_BATCH_POLICY_TARGET_SIZE,
+        ::metrics::Unit::Count,
+        "Target batch size chosen by batch policy."
+    );
+    ::metrics::describe_counter!(
+        METRICS_BATCH_POLICY_FORCE_SEND,
+        ::metrics::Unit::Count,
+        "Number of forced sends triggered by policy."
+    );
+    ::metrics::describe_counter!(
+        METRICS_BATCH_POLICY_DEFER,
+        ::metrics::Unit::Count,
+        "Number of policy deferrals by reason."
+    );
+}
+
+pub fn record_http_latency_ms(path: &str, status: u16, latency_ms: f64) {
+    let status_class = match status / 100 {
+        1 => "1xx",
+        2 => "2xx",
+        3 => "3xx",
+        4 => "4xx",
+        5 => "5xx",
+        _ => "other",
+    };
+
+    ::metrics::histogram!(
+        METRICS_HTTP_LATENCY_MS,
+        "route" => normalize_path(path),
+        "status_class" => status_class
+    )
+    .record(latency_ms);
+}
+
+pub fn increment_root_cache_hit() {
+    ::metrics::counter!(METRICS_ROOT_CACHE_HITS).increment(1);
+}
+
+pub fn increment_root_cache_miss() {
+    ::metrics::counter!(METRICS_ROOT_CACHE_MISSES).increment(1);
+}
+
+pub fn record_batch_submitted(batch_type: &'static str, batch_size: usize) {
+    ::metrics::counter!(METRICS_BATCH_SUBMITTED, "type" => batch_type).increment(1);
+    ::metrics::histogram!(METRICS_BATCH_SIZE, "type" => batch_type).record(batch_size as f64);
+}
+
+pub fn record_batch_result(batch_type: &'static str, success: bool, latency_ms: f64) {
+    ::metrics::histogram!(METRICS_BATCH_LATENCY_MS, "type" => batch_type).record(latency_ms);
+
+    if success {
+        ::metrics::counter!(METRICS_BATCH_SUCCESS, "type" => batch_type).increment(1);
+    } else {
+        ::metrics::counter!(METRICS_BATCH_FAILURE, "type" => batch_type).increment(1);
+    }
+}
+
+pub fn record_policy_scores(
+    batch_type: &'static str,
+    cost_score: f64,
+    urgency_score: f64,
+    target_batch_size: usize,
+) {
+    ::metrics::histogram!(METRICS_BATCH_POLICY_COST_SCORE, "type" => batch_type).record(cost_score);
+    ::metrics::histogram!(METRICS_BATCH_POLICY_URGENCY_SCORE, "type" => batch_type)
+        .record(urgency_score);
+    ::metrics::histogram!(METRICS_BATCH_POLICY_TARGET_SIZE, "type" => batch_type)
+        .record(target_batch_size as f64);
+}
+
+pub fn increment_policy_force_send(batch_type: &'static str) {
+    ::metrics::counter!(METRICS_BATCH_POLICY_FORCE_SEND, "type" => batch_type).increment(1);
+}
+
+pub fn increment_policy_defer(batch_type: &'static str, reason: &'static str) {
+    ::metrics::counter!(METRICS_BATCH_POLICY_DEFER, "type" => batch_type, "reason" => reason)
+        .increment(1);
+}
+
+fn normalize_path(path: &str) -> String {
+    // Replace dynamic segments like /status/{id} with /status/:id
+    if path.starts_with("/status/") {
+        return "/status/:id".to_string();
+    }
+    path.to_string()
+}

services/gateway/src/metrics.rs

@@ -9,10 +9,7 @@ use crate::{
     batch_policy::BaseFeeCache,
     config::BatchPolicyConfig,
     error::parse_contract_error,
-    metrics::{
-        METRICS_BATCH_FAILURE, METRICS_BATCH_LATENCY_MS, METRICS_BATCH_SIZE,
-        METRICS_BATCH_SUBMITTED, METRICS_BATCH_SUCCESS,
-    },
+    metrics,
     policy_batcher::{PolicyBatchLoopRunner, TimedEnvelope},
     request_tracker::BacklogScope,
 };
@@ -95,8 +92,7 @@ impl OpsBatcherRunner {
         let batch_size = batch.len();
         let ids: Vec<String> = batch.iter().map(|env| env.id.clone()).collect();
 
-        ::metrics::counter!(METRICS_BATCH_SUBMITTED, "type" => "ops").increment(1);
-        ::metrics::histogram!(METRICS_BATCH_SIZE, "type" => "ops").record(batch_size as f64);
+        metrics::record_batch_submitted("ops", batch_size);
 
         self.tracker
             .set_status_batch(&ids, GatewayRequestState::Batching)
@@ -116,8 +112,7 @@ impl OpsBatcherRunner {
         match res {
             Ok(builder) => {
                 let latency_ms = start.elapsed().as_millis() as f64;
-                ::metrics::histogram!(METRICS_BATCH_LATENCY_MS, "type" => "ops").record(latency_ms);
-                ::metrics::counter!(METRICS_BATCH_SUCCESS, "type" => "ops").increment(1);
+                metrics::record_batch_result("ops", true, latency_ms);
 
                 let hash = format!("0x{:x}", builder.tx_hash());
                 self.tracker
@@ -154,8 +149,7 @@ impl OpsBatcherRunner {
             }
             Err(e) => {
                 let latency_ms = start.elapsed().as_millis() as f64;
-                ::metrics::histogram!(METRICS_BATCH_LATENCY_MS, "type" => "ops").record(latency_ms);
-                ::metrics::counter!(METRICS_BATCH_FAILURE, "type" => "ops").increment(1);
+                metrics::record_batch_result("ops", false, latency_ms);
 
                 tracing::warn!(error = %e, "multicall3 send failed");
                 let error_str = e.to_string();

services/gateway/src/ops_batcher.rs

@@ -1,9 +1,4 @@
-use crate::{
-    error::GatewayErrorResponse,
-    metrics::{METRICS_ROOT_CACHE_HITS, METRICS_ROOT_CACHE_MISSES},
-    request::Registry,
-    types::AppState,
-};
+use crate::{error::GatewayErrorResponse, metrics, request::Registry, types::AppState};
 use alloy::primitives::U256;
 use axum::{Json, extract::State};
 use std::{
@@ -90,10 +85,10 @@ pub(crate) async fn is_valid_root(
 ) -> Result<Json<IsValidRootResponse>, GatewayErrorResponse> {
     let root = req_u256("root", &q.root)?;
     if is_cached_root(&state, root).await {
-        ::metrics::counter!(METRICS_ROOT_CACHE_HITS).increment(1);
+        metrics::increment_root_cache_hit();
         return Ok(Json(IsValidRootResponse { valid: true }));
     }
-    ::metrics::counter!(METRICS_ROOT_CACHE_MISSES).increment(1);
+    metrics::increment_root_cache_miss();
     let now = now_timestamp()?;
 
     let valid = state