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
|
use hyper::{
Response, Body, Request, Server,
header::{CONTENT_TYPE},
service::{make_service_fn, service_fn},
};
use std::net::ToSocketAddrs;
use prometheus::{Encoder, TextEncoder};
use log::{info,error};
use serde::Deserialize;
#[derive(Clone, Debug, Deserialize)]
/// Options for the monitoring service
pub struct MonitoringConfiguration {
enabled: bool,
address: Option<String>,
port: Option<i32>,
}
/// Handler for the hyper http server
async fn serve_metrics(_request: Request<Body>) -> Result<Response<Body>, hyper::Error> {
let encoder = TextEncoder::new();
let metrics = prometheus::gather();
let mut buffer = vec![];
encoder.encode(&metrics, &mut buffer).unwrap();
let response = Response::builder()
.status(200)
.header(CONTENT_TYPE, encoder.format_type())
.body(Body::from(buffer))
.unwrap();
Ok(response)
}
/// Starts a monitoring server on the requested port
pub fn start_monitoring(configuration: &MonitoringConfiguration) {
let config = configuration.clone();
tokio::task::spawn(async move {
if config.enabled {
let address = format!("{}:{}",
config.address.expect("a listening address must be specified for the metrics server"),
config.port.expect("a listening port must be specified for the metrics server")
);
info!("Starting monitoring server on {}", address);
let listen_address = address
.to_socket_addrs()
.unwrap()
.next()
.unwrap();
let server = Server::bind(&listen_address).serve(make_service_fn(|_| async {
Ok::<_, hyper::Error>(service_fn(serve_metrics))
}));
if let Err(e) = server.await {
error!("failed to start the monitoring server {}", e);
}
}
});
}
|
