openzeppelin_relayer/metrics/
mod.rs

1//! Metrics module for the application.
2//!
3//! - This module contains the global Prometheus registry.
4//! - Defines specific metrics for the application.
5
6pub mod middleware;
7use lazy_static::lazy_static;
8use prometheus::{
9    CounterVec, Encoder, Gauge, GaugeVec, HistogramOpts, HistogramVec, Opts, Registry, TextEncoder,
10};
11
12// Stage labels for TRANSACTION_PROCESSING_TIME histogram.
13pub const STAGE_REQUEST_QUEUE_DWELL: &str = "request_queue_dwell";
14pub const STAGE_PREPARE_DURATION: &str = "prepare_duration";
15pub const STAGE_SUBMISSION_QUEUE_DWELL: &str = "submission_queue_dwell";
16pub const STAGE_SUBMIT_DURATION: &str = "submit_duration";
17
18/// Observe a duration on the `TRANSACTION_PROCESSING_TIME` histogram.
19pub fn observe_processing_time(relayer_id: &str, network_type: &str, stage: &str, secs: f64) {
20    TRANSACTION_PROCESSING_TIME
21        .with_label_values(&[relayer_id, network_type, stage])
22        .observe(secs);
23}
24
25/// Observe queue pickup latency (time from send to consumer pickup).
26pub fn observe_queue_pickup_latency(queue_type: &str, backend: &str, secs: f64) {
27    QUEUE_PICKUP_LATENCY
28        .with_label_values(&[queue_type, backend])
29        .observe(secs);
30}
31
32/// Set the current backlog depth for a queue.
33///
34/// Labeled by `backend` (e.g. "pubsub") and `queue_type`. Fed by the backend's
35/// low-frequency, batched backlog read. Depth that is unavailable (e.g. under
36/// the emulator) is intentionally NOT reported here so it cannot read as 0.
37pub fn set_queue_depth(backend: &str, queue_type: &str, depth: f64) {
38    QUEUE_DEPTH
39        .with_label_values(&[backend, queue_type])
40        .set(depth);
41}
42
43/// Record the effective worker-thread count for a runtime at startup.
44///
45/// Labeled by `runtime`: "pipeline" (the multi-thread tokio runtime hosting the
46/// transaction pipeline) and "http" (the actix HTTP worker count). Backs the
47/// container-sizing observability contract (SC-003/SC-004).
48pub fn set_worker_threads(runtime: &str, count: usize) {
49    WORKER_THREADS
50        .with_label_values(&[runtime])
51        .set(count as f64);
52}
53use sysinfo::{Disks, System};
54
55lazy_static! {
56    // Global Prometheus registry.
57    pub static ref REGISTRY: Registry = Registry::new();
58
59    // Counter: Total HTTP requests.
60    pub static ref REQUEST_COUNTER: CounterVec = {
61        let opts = Opts::new("requests_total", "Total number of HTTP requests");
62        let counter_vec = CounterVec::new(opts, &["endpoint", "method", "status"]).unwrap();
63        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
64        counter_vec
65    };
66
67    // Counter: Total HTTP requests by raw URI.
68    pub static ref RAW_REQUEST_COUNTER: CounterVec = {
69      let opts = Opts::new("raw_requests_total", "Total number of HTTP requests by raw URI");
70      let counter_vec = CounterVec::new(opts, &["raw_uri", "method", "status"]).unwrap();
71      REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
72      counter_vec
73    };
74
75    // Histogram for request latency in seconds.
76    pub static ref REQUEST_LATENCY: HistogramVec = {
77      let histogram_opts = HistogramOpts::new("request_latency_seconds", "Request latency in seconds")
78          .buckets(vec![0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 25.0, 50.0, 100.0]);
79      let histogram_vec = HistogramVec::new(histogram_opts, &["endpoint", "method", "status"]).unwrap();
80      REGISTRY.register(Box::new(histogram_vec.clone())).unwrap();
81      histogram_vec
82    };
83
84    // Counter for error responses.
85    pub static ref ERROR_COUNTER: CounterVec = {
86        let opts = Opts::new("error_requests_total", "Total number of error responses");
87        // Using "status" to record the HTTP status code (or a special label like "service_error")
88        let counter_vec = CounterVec::new(opts, &["endpoint", "method", "status"]).unwrap();
89        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
90        counter_vec
91    };
92
93    // Gauge for CPU usage percentage.
94    pub static ref CPU_USAGE: Gauge = {
95      let gauge = Gauge::new("cpu_usage_percentage", "Current CPU usage percentage").unwrap();
96      REGISTRY.register(Box::new(gauge.clone())).unwrap();
97      gauge
98    };
99
100    // Gauge for memory usage percentage.
101    pub static ref MEMORY_USAGE_PERCENT: Gauge = {
102      let gauge = Gauge::new("memory_usage_percentage", "Memory usage percentage").unwrap();
103      REGISTRY.register(Box::new(gauge.clone())).unwrap();
104      gauge
105    };
106
107    // Gauge for memory usage in bytes.
108    pub static ref MEMORY_USAGE: Gauge = {
109        let gauge = Gauge::new("memory_usage_bytes", "Memory usage in bytes").unwrap();
110        REGISTRY.register(Box::new(gauge.clone())).unwrap();
111        gauge
112    };
113
114    // Gauge for total memory in bytes.
115    pub static ref TOTAL_MEMORY: Gauge = {
116      let gauge = Gauge::new("total_memory_bytes", "Total memory in bytes").unwrap();
117      REGISTRY.register(Box::new(gauge.clone())).unwrap();
118      gauge
119    };
120
121    // Gauge for available memory in bytes.
122    pub static ref AVAILABLE_MEMORY: Gauge = {
123        let gauge = Gauge::new("available_memory_bytes", "Available memory in bytes").unwrap();
124        REGISTRY.register(Box::new(gauge.clone())).unwrap();
125        gauge
126    };
127
128    // Gauge for used disk space in bytes.
129    pub static ref DISK_USAGE: Gauge = {
130      let gauge = Gauge::new("disk_usage_bytes", "Used disk space in bytes").unwrap();
131      REGISTRY.register(Box::new(gauge.clone())).unwrap();
132      gauge
133    };
134
135    // Gauge for disk usage percentage.
136    pub static ref DISK_USAGE_PERCENT: Gauge = {
137      let gauge = Gauge::new("disk_usage_percentage", "Disk usage percentage").unwrap();
138      REGISTRY.register(Box::new(gauge.clone())).unwrap();
139      gauge
140    };
141
142    // Gauge for in-flight requests.
143    pub static ref IN_FLIGHT_REQUESTS: GaugeVec = {
144        let gauge_vec = GaugeVec::new(
145            Opts::new("in_flight_requests", "Number of in-flight requests"),
146            &["endpoint"]
147        ).unwrap();
148        REGISTRY.register(Box::new(gauge_vec.clone())).unwrap();
149        gauge_vec
150    };
151
152    // Counter for request timeouts.
153    pub static ref TIMEOUT_COUNTER: CounterVec = {
154        let opts = Opts::new("request_timeouts_total", "Total number of request timeouts");
155        let counter_vec = CounterVec::new(opts, &["endpoint", "method", "timeout_type"]).unwrap();
156        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
157        counter_vec
158    };
159
160    // Gauge for file descriptor count.
161    pub static ref FILE_DESCRIPTORS: Gauge = {
162        let gauge = Gauge::new("file_descriptors_count", "Current file descriptor count").unwrap();
163        REGISTRY.register(Box::new(gauge.clone())).unwrap();
164        gauge
165    };
166
167    // Gauge for effective runtime worker-thread counts (labeled by runtime: "pipeline"/"http").
168    pub static ref WORKER_THREADS: GaugeVec = {
169        let gauge_vec = GaugeVec::new(
170            Opts::new("worker_threads", "Effective worker-thread count per runtime"),
171            &["runtime"]
172        ).unwrap();
173        REGISTRY.register(Box::new(gauge_vec.clone())).unwrap();
174        gauge_vec
175    };
176
177    // Gauge for CLOSE_WAIT socket count.
178    pub static ref CLOSE_WAIT_SOCKETS: Gauge = {
179        let gauge = Gauge::new("close_wait_sockets_count", "Number of CLOSE_WAIT sockets").unwrap();
180        REGISTRY.register(Box::new(gauge.clone())).unwrap();
181        gauge
182    };
183
184    // Counter for successful transactions (Confirmed status).
185    pub static ref TRANSACTIONS_SUCCESS: CounterVec = {
186        let opts = Opts::new("transactions_success_total", "Total number of successful transactions");
187        let counter_vec = CounterVec::new(opts, &["relayer_id", "network_type"]).unwrap();
188        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
189        counter_vec
190    };
191
192    // Counter for failed transactions (Failed, Expired, Canceled statuses).
193    // Labels: relayer_id, network_type, failure_reason, previous_status.
194    // Note: `previous_status` label added to track the pipeline stage before the failure
195    // (e.g. "pending", "sent", "submitted"), enabling pre- vs post-submission attribution.
196    pub static ref TRANSACTIONS_FAILED: CounterVec = {
197        let opts = Opts::new("transactions_failed_total", "Total number of failed transactions");
198        let counter_vec = CounterVec::new(opts, &["relayer_id", "network_type", "failure_reason", "previous_status"]).unwrap();
199        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
200        counter_vec
201    };
202
203    // Counter for RPC failures during API requests (before transaction creation).
204    // This tracks failures that occur during operations like get_status, get_balance, etc.
205    // that happen before a transaction is created.
206    pub static ref API_RPC_FAILURES: CounterVec = {
207        let opts = Opts::new("api_rpc_failures_total", "Total number of RPC failures during API requests (before transaction creation)");
208        let counter_vec = CounterVec::new(opts, &["relayer_id", "network_type", "operation_name", "error_type"]).unwrap();
209        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
210        counter_vec
211    };
212
213    // Counter for transaction creation (when a transaction is successfully created in the repository).
214    pub static ref TRANSACTIONS_CREATED: CounterVec = {
215        let opts = Opts::new("transactions_created_total", "Total number of transactions created");
216        let counter_vec = CounterVec::new(opts, &["relayer_id", "network_type"]).unwrap();
217        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
218        counter_vec
219    };
220
221    // Counter for transaction submissions (when status changes to Submitted).
222    pub static ref TRANSACTIONS_SUBMITTED: CounterVec = {
223        let opts = Opts::new("transactions_submitted_total", "Total number of transactions submitted to the network");
224        let counter_vec = CounterVec::new(opts, &["relayer_id", "network_type"]).unwrap();
225        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
226        counter_vec
227    };
228
229    // Gauge for transaction status distribution (current count of transactions in each status).
230    pub static ref TRANSACTIONS_BY_STATUS: GaugeVec = {
231        let gauge_vec = GaugeVec::new(
232            Opts::new("transactions_by_status", "Current number of transactions by status"),
233            &["relayer_id", "network_type", "status"]
234        ).unwrap();
235        REGISTRY.register(Box::new(gauge_vec.clone())).unwrap();
236        gauge_vec
237    };
238
239    // Histogram for transaction processing times (creation to submission).
240    pub static ref TRANSACTION_PROCESSING_TIME: HistogramVec = {
241        let histogram_opts = HistogramOpts::new("transaction_processing_seconds", "Transaction processing time in seconds")
242            .buckets(vec![0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0]);
243        let histogram_vec = HistogramVec::new(histogram_opts, &["relayer_id", "network_type", "stage"]).unwrap();
244        REGISTRY.register(Box::new(histogram_vec.clone())).unwrap();
245        histogram_vec
246    };
247
248    // Histogram for queue pickup latency (time from send to consumer pickup).
249    pub static ref QUEUE_PICKUP_LATENCY: HistogramVec = {
250        let histogram_opts = HistogramOpts::new("queue_pickup_latency_seconds", "Queue pickup latency in seconds (send to consumer pickup)")
251            .buckets(vec![0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, 600.0, 1800.0, 3600.0, 14400.0, 86400.0]);
252        let histogram_vec = HistogramVec::new(histogram_opts, &["queue_type", "backend"]).unwrap();
253        REGISTRY.register(Box::new(histogram_vec.clone())).unwrap();
254        histogram_vec
255    };
256
257    // Gauge for queue backlog depth. Labeled by backend and
258    // queue type. Currently emitted by the Pub/Sub backend (the first compliant
259    // backend); Redis/SQS do not yet populate it.
260    pub static ref QUEUE_DEPTH: GaugeVec = {
261        let opts = Opts::new("queue_depth", "Current queue backlog depth (undelivered messages)");
262        let gauge_vec = GaugeVec::new(opts, &["backend", "queue_type"]).unwrap();
263        REGISTRY.register(Box::new(gauge_vec.clone())).unwrap();
264        gauge_vec
265    };
266
267    // Histogram for RPC call latency.
268    pub static ref RPC_CALL_LATENCY: HistogramVec = {
269        let histogram_opts = HistogramOpts::new("rpc_call_latency_seconds", "RPC call latency in seconds")
270            .buckets(vec![0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0]);
271        let histogram_vec = HistogramVec::new(histogram_opts, &["relayer_id", "network_type", "operation_name"]).unwrap();
272        REGISTRY.register(Box::new(histogram_vec.clone())).unwrap();
273        histogram_vec
274    };
275
276    // Counter for Stellar transaction submission failures with decoded result codes.
277    pub static ref STELLAR_SUBMISSION_FAILURES: CounterVec = {
278        let opts = Opts::new("stellar_submission_failures_total",
279            "Stellar transaction submission failures by status and result code");
280        let counter_vec = CounterVec::new(opts, &["submit_status", "result_code"]).unwrap();
281        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
282        counter_vec
283    };
284
285    // Counter for plugin calls (tracks requests to /api/v1/plugins/{plugin_id}/call endpoints).
286    pub static ref PLUGIN_CALLS: CounterVec = {
287        let opts = Opts::new("plugin_calls_total", "Total number of plugin calls");
288        let counter_vec = CounterVec::new(opts, &["plugin_id", "method", "status"]).unwrap();
289        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
290        counter_vec
291    };
292
293    // Counter for Stellar submit responses with TRY_AGAIN_LATER status.
294    pub static ref STELLAR_TRY_AGAIN_LATER: CounterVec = {
295        let opts = Opts::new(
296            "stellar_try_again_later_total",
297            "Total number of Stellar transaction submit responses with TRY_AGAIN_LATER"
298        );
299        let counter_vec = CounterVec::new(opts, &["relayer_id", "tx_status"]).unwrap();
300        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
301        counter_vec
302    };
303
304    // Counter for transactions confirmed after experiencing TRY_AGAIN_LATER.
305    pub static ref TRANSACTIONS_TRY_AGAIN_LATER_SUCCESS: CounterVec = {
306        let opts = Opts::new(
307            "transactions_try_again_later_success_total",
308            "Total number of transactions confirmed after experiencing TRY_AGAIN_LATER"
309        );
310        let counter_vec = CounterVec::new(opts, &["relayer_id", "network_type"]).unwrap();
311        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
312        counter_vec
313    };
314
315    // Counter for transactions that failed after experiencing TRY_AGAIN_LATER.
316    pub static ref TRANSACTIONS_TRY_AGAIN_LATER_FAILED: CounterVec = {
317        let opts = Opts::new(
318            "transactions_try_again_later_failed_total",
319            "Total number of transactions that failed after experiencing TRY_AGAIN_LATER"
320        );
321        let counter_vec = CounterVec::new(opts, &["relayer_id", "network_type"]).unwrap();
322        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
323        counter_vec
324    };
325
326    // Counter for transactions that encountered an insufficient fee error.
327    pub static ref TRANSACTIONS_INSUFFICIENT_FEE: CounterVec = {
328        let opts = Opts::new(
329            "transactions_insufficient_fee_total",
330            "Total number of transactions that encountered an insufficient fee error"
331        );
332        let counter_vec = CounterVec::new(opts, &["relayer_id", "network_type"]).unwrap();
333        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
334        counter_vec
335    };
336
337    // Counter for transactions confirmed after experiencing insufficient fee.
338    pub static ref TRANSACTIONS_INSUFFICIENT_FEE_SUCCESS: CounterVec = {
339        let opts = Opts::new(
340            "transactions_insufficient_fee_success_total",
341            "Total number of transactions confirmed after experiencing insufficient fee"
342        );
343        let counter_vec = CounterVec::new(opts, &["relayer_id", "network_type"]).unwrap();
344        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
345        counter_vec
346    };
347
348    // Counter for transactions that failed after experiencing insufficient fee.
349    pub static ref TRANSACTIONS_INSUFFICIENT_FEE_FAILED: CounterVec = {
350        let opts = Opts::new(
351            "transactions_insufficient_fee_failed_total",
352            "Total number of transactions that failed after experiencing insufficient fee"
353        );
354        let counter_vec = CounterVec::new(opts, &["relayer_id", "network_type"]).unwrap();
355        REGISTRY.register(Box::new(counter_vec.clone())).unwrap();
356        counter_vec
357    };
358}
359
360/// Gather all metrics and encode into the provided format.
361pub fn gather_metrics() -> Result<Vec<u8>, Box<dyn std::error::Error>> {
362    let encoder = TextEncoder::new();
363    let metric_families = REGISTRY.gather();
364    let mut buffer = Vec::new();
365    encoder.encode(&metric_families, &mut buffer)?;
366    Ok(buffer)
367}
368
369/// Get file descriptor count for current process.
370fn get_fd_count() -> Result<usize, std::io::Error> {
371    let pid = std::process::id();
372
373    #[cfg(target_os = "linux")]
374    {
375        let fd_dir = format!("/proc/{pid}/fd");
376        std::fs::read_dir(fd_dir).map(|entries| entries.count())
377    }
378
379    #[cfg(target_os = "macos")]
380    {
381        use std::process::Command;
382        let output = Command::new("lsof")
383            .args(["-p", &pid.to_string()])
384            .output()?;
385        let count = String::from_utf8_lossy(&output.stdout)
386            .lines()
387            .count()
388            .saturating_sub(1); // Subtract header line
389        Ok(count)
390    }
391
392    #[cfg(not(any(target_os = "linux", target_os = "macos")))]
393    {
394        Ok(0) // Unsupported platform
395    }
396}
397
398/// Get CLOSE_WAIT socket count.
399fn get_close_wait_count() -> Result<usize, std::io::Error> {
400    #[cfg(any(target_os = "linux", target_os = "macos"))]
401    {
402        use std::process::Command;
403        let output = Command::new("sh")
404            .args(["-c", "netstat -an | grep CLOSE_WAIT | wc -l"])
405            .output()?;
406        let count = String::from_utf8_lossy(&output.stdout)
407            .trim()
408            .parse()
409            .unwrap_or(0);
410        Ok(count)
411    }
412
413    #[cfg(not(any(target_os = "linux", target_os = "macos")))]
414    {
415        Ok(0) // Unsupported platform
416    }
417}
418
419/// Updates the system metrics for CPU and memory usage.
420pub fn update_system_metrics() {
421    let mut sys = System::new_all();
422    sys.refresh_all();
423
424    // Overall CPU usage.
425    let cpu_usage = sys.global_cpu_usage();
426    CPU_USAGE.set(cpu_usage as f64);
427
428    // Total memory (in bytes).
429    let total_memory = sys.total_memory();
430    TOTAL_MEMORY.set(total_memory as f64);
431
432    // Available memory (in bytes).
433    let available_memory = sys.available_memory();
434    AVAILABLE_MEMORY.set(available_memory as f64);
435
436    // Used memory (in bytes).
437    let memory_usage = sys.used_memory();
438    MEMORY_USAGE.set(memory_usage as f64);
439
440    // Calculate memory usage percentage
441    let memory_percentage = if total_memory > 0 {
442        (memory_usage as f64 / total_memory as f64) * 100.0
443    } else {
444        0.0
445    };
446    MEMORY_USAGE_PERCENT.set(memory_percentage);
447
448    // Calculate disk usage:
449    // Sum total space and available space across all disks.
450    let disks = Disks::new_with_refreshed_list();
451    let mut total_disk_space: u64 = 0;
452    let mut total_disk_available: u64 = 0;
453    for disk in disks.list() {
454        total_disk_space += disk.total_space();
455        total_disk_available += disk.available_space();
456    }
457    // Used disk space is total minus available ( in bytes).
458    let used_disk_space = total_disk_space.saturating_sub(total_disk_available);
459    DISK_USAGE.set(used_disk_space as f64);
460
461    // Calculate disk usage percentage.
462    let disk_percentage = if total_disk_space > 0 {
463        (used_disk_space as f64 / total_disk_space as f64) * 100.0
464    } else {
465        0.0
466    };
467    DISK_USAGE_PERCENT.set(disk_percentage);
468
469    // Update file descriptor count.
470    if let Ok(fd_count) = get_fd_count() {
471        FILE_DESCRIPTORS.set(fd_count as f64);
472    }
473
474    // Update CLOSE_WAIT socket count.
475    if let Ok(close_wait) = get_close_wait_count() {
476        CLOSE_WAIT_SOCKETS.set(close_wait as f64);
477    }
478}
479
480#[cfg(test)]
481mod actix_tests {
482    use super::*;
483    use actix_web::{
484        dev::{Service, ServiceRequest, ServiceResponse, Transform},
485        http, test, Error, HttpResponse,
486    };
487    use futures::future::{self};
488    use middleware::MetricsMiddleware;
489    use prometheus::proto::MetricFamily;
490    use std::{
491        pin::Pin,
492        task::{Context, Poll},
493    };
494
495    // Dummy service that always returns a successful response (HTTP 200 OK).
496    struct DummySuccessService;
497
498    impl Service<ServiceRequest> for DummySuccessService {
499        type Response = ServiceResponse;
500        type Error = Error;
501        type Future = Pin<Box<dyn future::Future<Output = Result<Self::Response, Self::Error>>>>;
502
503        fn poll_ready(&self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
504            Poll::Ready(Ok(()))
505        }
506
507        fn call(&self, req: ServiceRequest) -> Self::Future {
508            let resp = req.into_response(HttpResponse::Ok().finish());
509            Box::pin(async move { Ok(resp) })
510        }
511    }
512
513    // Dummy service that always returns an error.
514    struct DummyErrorService;
515
516    impl Service<ServiceRequest> for DummyErrorService {
517        type Response = ServiceResponse;
518        type Error = Error;
519        type Future = Pin<Box<dyn future::Future<Output = Result<Self::Response, Self::Error>>>>;
520
521        fn poll_ready(&self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
522            Poll::Ready(Ok(()))
523        }
524
525        fn call(&self, _req: ServiceRequest) -> Self::Future {
526            Box::pin(async move { Err(actix_web::error::ErrorInternalServerError("dummy error")) })
527        }
528    }
529
530    // Helper function to find a metric family by name.
531    fn find_metric_family<'a>(
532        name: &str,
533        families: &'a [MetricFamily],
534    ) -> Option<&'a MetricFamily> {
535        families.iter().find(|mf| mf.name() == name)
536    }
537
538    #[actix_rt::test]
539    async fn test_gather_metrics_contains_expected_names() {
540        // Update system metrics
541        update_system_metrics();
542
543        // Increment request counters to ensure they appear in output
544        REQUEST_COUNTER
545            .with_label_values(&["/test", "GET", "200"])
546            .inc();
547        RAW_REQUEST_COUNTER
548            .with_label_values(&["/test?param=value", "GET", "200"])
549            .inc();
550        REQUEST_LATENCY
551            .with_label_values(&["/test", "GET", "200"])
552            .observe(0.1);
553        ERROR_COUNTER
554            .with_label_values(&["/test", "GET", "500"])
555            .inc();
556
557        // Touch insufficient fee metrics to ensure they appear in output
558        TRANSACTIONS_INSUFFICIENT_FEE
559            .with_label_values(&["test-relayer", "stellar"])
560            .inc();
561        TRANSACTIONS_INSUFFICIENT_FEE_SUCCESS
562            .with_label_values(&["test-relayer", "stellar"])
563            .inc();
564        TRANSACTIONS_INSUFFICIENT_FEE_FAILED
565            .with_label_values(&["test-relayer", "stellar"])
566            .inc();
567
568        // Touch TRY_AGAIN_LATER metrics to ensure they appear in output
569        TRANSACTIONS_TRY_AGAIN_LATER_SUCCESS
570            .with_label_values(&["test-relayer", "stellar"])
571            .inc();
572        TRANSACTIONS_TRY_AGAIN_LATER_FAILED
573            .with_label_values(&["test-relayer", "stellar"])
574            .inc();
575
576        // Queue pickup latency
577        observe_queue_pickup_latency("transaction-request", "sqs", 0.5);
578
579        // Queue depth gauge
580        set_queue_depth("pubsub", "status-check-evm", 7.0);
581
582        let metrics = gather_metrics().expect("failed to gather metrics");
583        let output = String::from_utf8(metrics).expect("metrics output is not valid UTF-8");
584
585        // System metrics
586        assert!(output.contains("cpu_usage_percentage"));
587        assert!(output.contains("memory_usage_percentage"));
588        assert!(output.contains("memory_usage_bytes"));
589        assert!(output.contains("total_memory_bytes"));
590        assert!(output.contains("available_memory_bytes"));
591        assert!(output.contains("disk_usage_bytes"));
592        assert!(output.contains("disk_usage_percentage"));
593
594        // Request metrics
595        assert!(output.contains("requests_total"));
596        assert!(output.contains("raw_requests_total"));
597        assert!(output.contains("request_latency_seconds"));
598        assert!(output.contains("error_requests_total"));
599
600        // Insufficient fee metrics
601        assert!(output.contains("transactions_insufficient_fee_total"));
602        assert!(output.contains("transactions_insufficient_fee_success_total"));
603        assert!(output.contains("transactions_insufficient_fee_failed_total"));
604
605        // TRY_AGAIN_LATER metrics
606        assert!(output.contains("transactions_try_again_later_success_total"));
607        assert!(output.contains("transactions_try_again_later_failed_total"));
608
609        // Queue pickup latency
610        assert!(output.contains("queue_pickup_latency_seconds"));
611
612        // Queue depth gauge
613        assert!(output.contains("queue_depth"));
614        assert!(output.contains(r#"queue_depth{backend="pubsub",queue_type="status-check-evm"} 7"#));
615    }
616
617    #[actix_rt::test]
618    async fn test_update_system_metrics() {
619        // Reset metrics to ensure clean state
620        CPU_USAGE.set(0.0);
621        TOTAL_MEMORY.set(0.0);
622        AVAILABLE_MEMORY.set(0.0);
623        MEMORY_USAGE.set(0.0);
624        MEMORY_USAGE_PERCENT.set(0.0);
625        DISK_USAGE.set(0.0);
626        DISK_USAGE_PERCENT.set(0.0);
627
628        // Call the function we're testing
629        update_system_metrics();
630
631        // Verify that metrics have been updated with reasonable values
632        let cpu_usage = CPU_USAGE.get();
633        assert!(
634            (0.0..=100.0).contains(&cpu_usage),
635            "CPU usage should be between 0-100%, got {cpu_usage}"
636        );
637
638        let memory_usage = MEMORY_USAGE.get();
639        assert!(
640            memory_usage >= 0.0,
641            "Memory usage should be >= 0, got {memory_usage}"
642        );
643
644        let memory_percent = MEMORY_USAGE_PERCENT.get();
645        assert!(
646            (0.0..=100.0).contains(&memory_percent),
647            "Memory usage percentage should be between 0-100%, got {memory_percent}"
648        );
649
650        let total_memory = TOTAL_MEMORY.get();
651        assert!(
652            total_memory > 0.0,
653            "Total memory should be > 0, got {total_memory}"
654        );
655
656        let available_memory = AVAILABLE_MEMORY.get();
657        assert!(
658            available_memory >= 0.0,
659            "Available memory should be >= 0, got {available_memory}"
660        );
661
662        let disk_usage = DISK_USAGE.get();
663        assert!(
664            disk_usage >= 0.0,
665            "Disk usage should be >= 0, got {disk_usage}"
666        );
667
668        let disk_percent = DISK_USAGE_PERCENT.get();
669        assert!(
670            (0.0..=100.0).contains(&disk_percent),
671            "Disk usage percentage should be between 0-100%, got {disk_percent}"
672        );
673
674        // Verify that memory usage doesn't exceed total memory
675        assert!(
676            memory_usage <= total_memory,
677            "Memory usage should be <= total memory, got {memory_usage}"
678        );
679
680        // Verify that available memory plus used memory doesn't exceed total memory
681        assert!(
682            (available_memory + memory_usage) <= total_memory,
683            "Available memory plus used memory should be <= total memory {}, got {}",
684            total_memory,
685            available_memory + memory_usage
686        );
687    }
688
689    #[actix_rt::test]
690    async fn test_middleware_success() {
691        let req = test::TestRequest::with_uri("/test_success").to_srv_request();
692
693        let middleware = MetricsMiddleware;
694        let service = middleware.new_transform(DummySuccessService).await.unwrap();
695
696        let resp = service.call(req).await.unwrap();
697        assert_eq!(resp.response().status(), http::StatusCode::OK);
698
699        let families = REGISTRY.gather();
700        let counter_fam = find_metric_family("requests_total", &families)
701            .expect("requests_total metric family not found");
702
703        let mut found = false;
704        for m in counter_fam.get_metric() {
705            let labels = m.get_label();
706            if labels
707                .iter()
708                .any(|l| l.name() == "endpoint" && l.value() == "/test_success")
709            {
710                found = true;
711                assert!(m.get_counter().value() >= 1.0);
712            }
713        }
714        assert!(
715            found,
716            "Expected metric with endpoint '/test_success' not found"
717        );
718    }
719
720    #[actix_rt::test]
721    async fn test_middleware_error() {
722        let req = test::TestRequest::with_uri("/test_error").to_srv_request();
723
724        let middleware = MetricsMiddleware;
725        let service = middleware.new_transform(DummyErrorService).await.unwrap();
726
727        let result = service.call(req).await;
728        assert!(result.is_err());
729
730        let families = REGISTRY.gather();
731        let error_counter_fam = find_metric_family("error_requests_total", &families)
732            .expect("error_requests_total metric family not found");
733
734        let mut found = false;
735        for m in error_counter_fam.get_metric() {
736            let labels = m.get_label();
737            if labels
738                .iter()
739                .any(|l| l.name() == "endpoint" && l.value() == "/test_error")
740            {
741                found = true;
742                assert!(m.get_counter().value() >= 1.0);
743            }
744        }
745        assert!(
746            found,
747            "Expected error metric with endpoint '/test_error' not found"
748        );
749    }
750}
751
752#[cfg(test)]
753mod property_tests {
754    use proptest::{prelude::*, test_runner::Config};
755
756    // A helper function to compute percentage used from total.
757    fn compute_percentage(used: u64, total: u64) -> f64 {
758        if total > 0 {
759            (used as f64 / total as f64) * 100.0
760        } else {
761            0.0
762        }
763    }
764
765    proptest! {
766        // Set the number of cases to 1000
767        #![proptest_config(Config {
768          cases: 1000, ..Config::default()
769        })]
770
771        #[test]
772        fn prop_compute_percentage((total, used) in {
773            (1u64..1_000_000u64).prop_flat_map(|total| {
774                (Just(total), 0u64..=total)
775            })
776        }) {
777            let percentage = compute_percentage(used, total);
778            prop_assert!(percentage >= 0.0);
779            prop_assert!(percentage <= 100.0);
780        }
781
782        #[test]
783        fn prop_labels_are_reasonable(
784              endpoint in ".*",
785              method in prop::sample::select(vec![
786                "GET".to_string(),
787                "POST".to_string(),
788                "PUT".to_string(),
789                "DELETE".to_string()
790                ])
791            ) {
792            let endpoint_label = if endpoint.is_empty() { "/".to_string() } else { endpoint.clone() };
793            let method_label = method;
794
795            prop_assert!(endpoint_label.chars().count() <= 1024, "Endpoint label too long");
796            prop_assert!(method_label.chars().count() <= 16, "Method label too long");
797
798            let status = "200".to_string();
799            let labels = vec![endpoint_label, method_label, status];
800
801            for label in labels {
802                prop_assert!(!label.is_empty());
803                prop_assert!(label.len() < 1024);
804            }
805        }
806    }
807}
808
809#[cfg(test)]
810mod processing_time_tests {
811    use super::*;
812
813    #[test]
814    fn test_observe_processing_time_records_to_histogram() {
815        let before = TRANSACTION_PROCESSING_TIME
816            .with_label_values(&["test-relayer", "evm", "request_queue_dwell"])
817            .get_sample_count();
818
819        observe_processing_time("test-relayer", "evm", "request_queue_dwell", 1.5);
820
821        let after = TRANSACTION_PROCESSING_TIME
822            .with_label_values(&["test-relayer", "evm", "request_queue_dwell"])
823            .get_sample_count();
824
825        assert_eq!(after, before + 1, "sample count should increase by 1");
826    }
827
828    #[test]
829    fn test_observe_processing_time_accumulates_sum() {
830        let label = "test_sum_stage";
831        let before_sum = TRANSACTION_PROCESSING_TIME
832            .with_label_values(&["test-relayer-sum", "stellar", label])
833            .get_sample_sum();
834
835        observe_processing_time("test-relayer-sum", "stellar", label, 2.0);
836        observe_processing_time("test-relayer-sum", "stellar", label, 3.0);
837
838        let after_sum = TRANSACTION_PROCESSING_TIME
839            .with_label_values(&["test-relayer-sum", "stellar", label])
840            .get_sample_sum();
841
842        let delta = after_sum - before_sum;
843        assert!(
844            (delta - 5.0).abs() < 0.001,
845            "sum should increase by 5.0, got delta {delta}"
846        );
847    }
848
849    #[test]
850    fn test_stage_constants_are_distinct() {
851        let stages = [
852            STAGE_REQUEST_QUEUE_DWELL,
853            STAGE_PREPARE_DURATION,
854            STAGE_SUBMISSION_QUEUE_DWELL,
855            STAGE_SUBMIT_DURATION,
856        ];
857        let unique: std::collections::HashSet<&str> = stages.iter().copied().collect();
858        assert_eq!(stages.len(), unique.len(), "stage constants must be unique");
859    }
860
861    #[test]
862    fn test_observe_queue_pickup_latency_records_to_histogram() {
863        let before = QUEUE_PICKUP_LATENCY
864            .with_label_values(&["notification", "sqs"])
865            .get_sample_count();
866
867        observe_queue_pickup_latency("notification", "sqs", 1.5);
868
869        let after = QUEUE_PICKUP_LATENCY
870            .with_label_values(&["notification", "sqs"])
871            .get_sample_count();
872
873        assert_eq!(after, before + 1, "sample count should increase by 1");
874    }
875
876    #[test]
877    fn test_observe_queue_pickup_latency_both_backends() {
878        for backend in &["sqs", "redis"] {
879            let before = QUEUE_PICKUP_LATENCY
880                .with_label_values(&["relayer-health-check", backend])
881                .get_sample_count();
882
883            observe_queue_pickup_latency("relayer-health-check", backend, 0.25);
884
885            let after = QUEUE_PICKUP_LATENCY
886                .with_label_values(&["relayer-health-check", backend])
887                .get_sample_count();
888
889            assert_eq!(
890                after,
891                before + 1,
892                "sample count should increase by 1 for backend {backend}"
893            );
894        }
895    }
896}
openzeppelin_relayer/metrics/mod.rs

openzeppelin_relayer/metrics/
mod.rs
