rediq 0.2.4

A distributed task queue framework for Rust based on Redis
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

//! HTTP metrics server
//!
//! Provides an HTTP endpoint for Prometheus metrics scraping.

#[cfg(feature = "metrics-http")]
use super::RediqMetrics;
use std::net::SocketAddr;
#[cfg(feature = "metrics-http")]
use std::sync::Arc;
#[cfg(feature = "metrics-http")]
use warp::Filter;

/// HTTP metrics server
///
/// Serves Prometheus metrics on `/metrics` endpoint.
#[cfg(feature = "metrics-http")]
pub struct MetricsServer {
    metrics: Arc<RediqMetrics>,
    bind_address: SocketAddr,
}

#[cfg(feature = "metrics-http")]
impl MetricsServer {
    /// Create a new metrics server
    pub fn new(metrics: Arc<RediqMetrics>, bind_address: SocketAddr) -> Self {
        Self {
            metrics,
            bind_address,
        }
    }

    /// Create a new metrics server binding to 0.0.0.0:9090
    pub fn new_default(metrics: Arc<RediqMetrics>) -> Self {
        Self::new(
            metrics,
            SocketAddr::from(([0, 0, 0, 0], 9090))
        )
    }

    /// Get the metrics as a string
    pub fn get_metrics(&self) -> String {
        let encoder = prometheus::TextEncoder::new();
        let metric_families = self.metrics.registry().gather();
        let result = encoder.encode_to_string(&metric_families);
        match result {
            Ok(s) => s,
            Err(_) => String::from("# Error encoding metrics\n"),
        }
    }

    /// Start the HTTP server
    pub async fn run(self) -> Result<(), crate::Error> {
        let metrics = self.metrics.clone();

        // Build a simple warp filter that returns metrics as plain text
        let route = warp::path!("metrics")
            .and(warp::get())
            .map(move || {
                let encoder = prometheus::TextEncoder::new();
                let metric_families = metrics.registry().gather();
                let result = encoder.encode_to_string(&metric_families);
                let output = match result {
                    Ok(s) => s,
                    Err(_) => String::from("# Error encoding metrics\n"),
                };
                warp::reply::with_header(output, "content-type", "text/plain")
            });

        let addr = self.bind_address;
        tracing::info!("Metrics server listening on http://{}", addr);
        tracing::info!("Metrics available at http://{}/metrics", addr);

        warp::serve(route).run(addr).await;
        Ok(())
    }

    /// Start the HTTP server in the background
    pub fn spawn(self) -> tokio::task::JoinHandle<Result<(), crate::Error>> {
        tokio::spawn(async move {
            self.run().await
        })
    }
}

#[cfg(not(feature = "metrics-http"))]
/// Metrics server is only available with the `metrics-http` feature
///
/// This is a placeholder type when the `metrics-http` feature is disabled.
pub struct MetricsServer {}

#[cfg(not(feature = "metrics-http"))]
impl MetricsServer {
    /// Create a new metrics server (only available with metrics-http feature)
    ///
    /// This function will panic if the `metrics-http` feature is not enabled.
    pub fn new(_metrics: std::sync::Arc<super::RediqMetrics>, _bind_address: SocketAddr) -> Self {
        panic!("MetricsServer requires the `metrics-http` feature to be enabled");
    }
}

#[cfg(test)]
mod tests {
    #[cfg(feature = "metrics-http")]
    #[test]
    fn test_bind_address() {
        use super::*;
        use super::super::RediqMetrics;
        use std::sync::Arc;

        let metrics = Arc::new(RediqMetrics::new_or_default());
        let server = MetricsServer::new_default(metrics);
        assert_eq!(server.bind_address.port(), 9090);
    }
}