Bump the hyper crate to v1.4.1 and rework prometheus server handling (#778)

sarajmunjal · web-flow · commit 7cbc9178d8ca · 2024-08-29T09:47:58.000-05:00
Bump hyper to v1.4.1 and rework prometheus server handling
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -29,7 +29,9 @@ base64 = "0.21"
 stringprep = "0.1"
 tokio-rustls = "0.24"
 rustls-pemfile = "1"
-hyper = { version = "0.14", features = ["full"] }
+http-body-util = "0.1.2"
+hyper = { version = "1.4.1", features = ["full"] }
+hyper-util = { version = "0.1.7", features = ["tokio"] }
 phf = { version = "0.11.1", features = ["macros"] }
 exitcode = "1.1.2"
 futures = "0.3"
diff --git a/src/prometheus.rs b/src/prometheus.rs
@@ -1,12 +1,19 @@
-use hyper::service::{make_service_fn, service_fn};
-use hyper::{Body, Method, Request, Response, Server, StatusCode};
+use http_body_util::Full;
+use hyper::body;
+use hyper::body::Bytes;
+
+use hyper::server::conn::http1;
+use hyper::service::service_fn;
+use hyper::{Method, Request, Response, StatusCode};
+use hyper_util::rt::TokioIo;
 use log::{debug, error, info};
 use phf::phf_map;
 use std::collections::HashMap;
 use std::fmt;
 use std::net::SocketAddr;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
+use tokio::net::TcpListener;
 
 use crate::config::Address;
 use crate::pool::{get_all_pools, PoolIdentifier};
@@ -243,7 +250,9 @@ impl<Value: fmt::Display> PrometheusMetric<Value> {
     }
 }
 
-async fn prometheus_stats(request: Request<Body>) -> Result<Response<Body>, hyper::http::Error> {
+async fn prometheus_stats(
+    request: Request<body::Incoming>,
+) -> Result<Response<Full<Bytes>>, hyper::http::Error> {
     match (request.method(), request.uri().path()) {
         (&Method::GET, "/metrics") => {
             let mut lines = Vec::new();
@@ -374,14 +383,35 @@ fn push_server_stats(lines: &mut Vec<String>) {
 }
 
 pub async fn start_metric_server(http_addr: SocketAddr) {
-    let http_service_factory =
-        make_service_fn(|_conn| async { Ok::<_, hyper::Error>(service_fn(prometheus_stats)) });
-    let server = Server::bind(&http_addr).serve(http_service_factory);
+    let listener = TcpListener::bind(http_addr);
+    let listener = match listener.await {
+        Ok(listener) => listener,
+        Err(e) => {
+            error!("Failed to bind prometheus server to HTTP address: {}.", e);
+            return;
+        }
+    };
     info!(
         "Exposing prometheus metrics on http://{}/metrics.",
         http_addr
     );
-    if let Err(e) = server.await {
-        error!("Failed to run HTTP server: {}.", e);
+    loop {
+        let stream = match listener.accept().await {
+            Ok((stream, _)) => stream,
+            Err(e) => {
+                error!("Error accepting connection: {}", e);
+                continue;
+            }
+        };
+        let io = TokioIo::new(stream);
+
+        tokio::task::spawn(async move {
+            if let Err(err) = http1::Builder::new()
+                .serve_connection(io, service_fn(prometheus_stats))
+                .await
+            {
+                eprintln!("Error serving HTTP connection for metrics: {:?}", err);
+            }
+        });
     }
 }
