-
Notifications
You must be signed in to change notification settings - Fork 160
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Adds Proof Of Concept HTTP unix domain socket listener
- Loading branch information
1 parent
f84efc4
commit c4c4848
Showing
5 changed files
with
228 additions
and
5 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
62 changes: 62 additions & 0 deletions
62
metrics-exporter-prometheus/examples/prometheus_uds_server.rs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| use std::thread; | ||
| use std::time::Duration; | ||
|
|
||
| use metrics::{counter, describe_counter, describe_histogram, gauge, histogram}; | ||
| use metrics_exporter_prometheus::PrometheusBuilder; | ||
| use metrics_util::MetricKindMask; | ||
|
|
||
| use quanta::Clock; | ||
| use rand::{thread_rng, Rng}; | ||
|
|
||
| fn main() { | ||
| tracing_subscriber::fmt::init(); | ||
|
|
||
| let builder = PrometheusBuilder::new().with_http_uds_listener("/tmp/metrics.sock"); | ||
| builder | ||
| .idle_timeout( | ||
| MetricKindMask::COUNTER | MetricKindMask::HISTOGRAM, | ||
| Some(Duration::from_secs(10)), | ||
| ) | ||
| .install() | ||
| .expect("failed to install Prometheus recorder"); | ||
|
|
||
| // We register these metrics, which gives us a chance to specify a description for them. The | ||
| // Prometheus exporter records this description and adds it as HELP text when the endpoint is | ||
| // scraped. | ||
| // | ||
| // Registering metrics ahead of using them is not required, but is the only way to specify the | ||
| // description of a metric. | ||
| describe_counter!("tcp_server_loops", "The iterations of the TCP server event loop so far."); | ||
| describe_histogram!( | ||
| "tcp_server_loop_delta_secs", | ||
| "The time taken for iterations of the TCP server event loop." | ||
| ); | ||
|
|
||
| let clock = Clock::new(); | ||
| let mut last = None; | ||
|
|
||
| counter!("idle_metric").increment(1); | ||
| gauge!("testing").set(42.0); | ||
|
|
||
| // Loop over and over, pretending to do some work. | ||
| loop { | ||
| counter!("tcp_server_loops", "system" => "foo").increment(1); | ||
|
|
||
| if let Some(t) = last { | ||
| let delta: Duration = clock.now() - t; | ||
| histogram!("tcp_server_loop_delta_secs", "system" => "foo").record(delta); | ||
| } | ||
|
|
||
| let increment_gauge = thread_rng().gen_bool(0.75); | ||
| let gauge = gauge!("lucky_iterations"); | ||
| if increment_gauge { | ||
| gauge.increment(1.0); | ||
| } else { | ||
| gauge.decrement(1.0); | ||
| } | ||
|
|
||
| last = Some(clock.now()); | ||
|
|
||
| thread::sleep(Duration::from_millis(750)); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,96 @@ | ||
| use std::net::SocketAddr; | ||
|
|
||
| use http_body_util::Full; | ||
| use hyper::{ | ||
| body::{self, Bytes, Incoming}, | ||
| server::conn::http1::Builder as HyperHttpBuilder, | ||
| service::service_fn, | ||
| Request, Response, StatusCode, | ||
| }; | ||
| use hyper_util::rt::TokioIo; | ||
| use ipnet::IpNet; | ||
| use std::path::PathBuf; | ||
| use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream}; | ||
| use tracing::warn; | ||
|
|
||
| use crate::{common::BuildError, ExporterFuture, PrometheusHandle}; | ||
|
|
||
| struct UnixListeningExporter { | ||
| handle: PrometheusHandle, | ||
| } | ||
|
|
||
| impl UnixListeningExporter { | ||
| async fn serve(&self, listener: UnixListener) -> Result<(), hyper::Error> { | ||
| loop { | ||
| let stream = match listener.accept().await { | ||
| Ok((stream, _)) => stream, | ||
| Err(e) => { | ||
| warn!(error = ?e, "Error accepting connection. Ignoring request."); | ||
| continue; | ||
| } | ||
| }; | ||
|
|
||
| self.process_stream(stream).await; | ||
| } | ||
| } | ||
|
|
||
| async fn process_stream(&self, stream: UnixStream) { | ||
| let handle = self.handle.clone(); | ||
| let service = service_fn(move |req: Request<body::Incoming>| { | ||
| let handle = handle.clone(); | ||
| async move { Ok::<_, hyper::Error>(Self::handle_http_request(&handle, &req)) } | ||
| }); | ||
|
|
||
| tokio::spawn(async move { | ||
| if let Err(err) = | ||
| HyperHttpBuilder::new().serve_connection(TokioIo::new(stream), service).await | ||
| { | ||
| warn!(error = ?err, "Error serving connection."); | ||
| }; | ||
| }); | ||
| } | ||
|
|
||
| fn handle_http_request( | ||
| handle: &PrometheusHandle, | ||
| req: &Request<Incoming>, | ||
| ) -> Response<Full<Bytes>> { | ||
| Response::new(match req.uri().path() { | ||
| "/health" => "OK".into(), | ||
| _ => handle.render().into(), | ||
| }) | ||
| } | ||
|
|
||
| fn new_forbidden_response() -> Response<Full<Bytes>> { | ||
| // This unwrap should not fail because we don't use any function that | ||
| // can assign an Err to it's inner such as `Builder::header``. A unit test | ||
| // will have to suffice to detect if this fails to hold true. | ||
| Response::builder().status(StatusCode::FORBIDDEN).body(Full::<Bytes>::default()).unwrap() | ||
| } | ||
| } | ||
|
|
||
| /// Creates an `ExporterFuture` implementing a http listener that servies prometheus metrics. | ||
| /// | ||
| /// # Errors | ||
| /// Will return Err if it cannot bind to the listen address | ||
| pub(crate) fn new_http_uds_listener( | ||
| handle: PrometheusHandle, | ||
| listen_path: PathBuf, | ||
| ) -> Result<ExporterFuture, BuildError> { | ||
| let listener = UnixListener::bind(listen_path) | ||
| .and_then(|listener| Ok(listener)) | ||
| .map_err(|e| BuildError::FailedToCreateHTTPListener(e.to_string()))?; | ||
|
|
||
| let exporter = UnixListeningExporter { handle }; | ||
|
|
||
| Ok(Box::pin(async move { exporter.serve(listener).await })) | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| mod tests { | ||
| use crate::exporter::uds_listener::UnixListeningExporter; | ||
|
|
||
| #[test] | ||
| fn new_forbidden_response_always_succeeds() { | ||
| UnixListeningExporter::new_forbidden_response(); // doesn't panic | ||
| } | ||
| } |