Update crates (#977)

Unfortunately this isn't as big of a win as I would have hoped, primarily due to tonic and hickory-resolver both using many outdated crates, but it is a step in the right direction.

- hyper -> 1.3 - This aligns better with some crates eg. kube-client, but misaligns us with tonic/axum. This is also annoying because 1.0 re/moved a lot of functionality so we have new dependencies on small utility crates, but it is what it is
- trust-dns-resolver -> hickory-resolver - trust-dns-resolver was a major source of outdated crates, hickory-resolver, being a fork of that project, carries on the tradition, though is _slightly_ better, as well as just being the future as opposed to trust-dns

Cargo.toml

@@ -85,8 +85,12 @@ enum-map.workspace = true
 eyre.workspace = true
 fixedstr.workspace = true
 futures.workspace = true
-hyper = { version = "0.14.27", features = ["http2"] }
-hyper-rustls = { version = "0.24.1", features = ["http2", "webpki-roots"] }
+http-body-util = "0.1"
+hyper = { version = "1.3", features = ["http2", "http1", "server"] }
+hyper-rustls = { version = "0.27", default-features = false, features = [
+    "http2",
+    "webpki-roots",
+] }
 ipnetwork = "0.20.0"
 k8s-openapi.workspace = true
 lz4_flex = { version = "0.11", default-features = false }
@@ -114,6 +118,7 @@ thiserror.workspace = true
 tokio.workspace = true
 tokio-stream.workspace = true
 tonic.workspace = true
+tower.workspace = true
 tracing.workspace = true
 tracing-futures.workspace = true
 tracing-subscriber = { workspace = true, features = ["json", "env-filter"] }
@@ -123,21 +128,24 @@ uuid.workspace = true
 lasso = { version = "0.7.2", features = ["multi-threaded"] }
 kube.workspace = true
 kube-core.workspace = true
-trust-dns-resolver = { version = "0.23.0", features = [
-    "tokio",
-    "tokio-rustls",
+hickory-resolver = { version = "0.24", features = [
     "dns-over-https-rustls",
+    "system-config",
 ] }
 async-trait = "0.1.73"
-strum = "0.25.0"
-strum_macros = "0.25.2"
+strum = "0.26"
+strum_macros = "0.26"
 cfg-if = "1.0.0"
 libflate = "2.0.0"
 form_urlencoded = "1.2.1"
 
+[dependencies.hyper-util]
+version = "0.1"
+features = ["client", "client-legacy"]
+
 [target.'cfg(target_os = "linux")'.dependencies]
 sys-info = "0.9.1"
-tokio-uring = { version = "0.4.0", features = ["bytes"] }
+tokio-uring = { version = "0.5", features = ["bytes"] }
 pprof = { version = "0.13.0", features = ["prost", "prost-codec"] }
 
 [dev-dependencies]
@@ -150,13 +158,6 @@ tracing-test = "0.2.4"
 tempfile.workspace = true
 xxhash-rust = { version = "0.8", features = ["xxh3"] }
 
-[build-dependencies]
-tonic-build = { version = "0.10.2", default_features = false, features = [
-    "transport",
-    "prost",
-] }
-prost-build = "0.12.1"
-
 [features]
 default = []
 instrument = []
@@ -177,8 +178,8 @@ edition = "2021"
 arc-swap = { version = "1.6.0", features = ["serde"] }
 async-channel = "2.1.0"
 async-stream = "0.3.5"
-base64 = "0.21.0"
-cached = { version = "0.49", default-features = false }
+base64 = "0.22.0"
+cached = { version = "0.51", default-features = false }
 eyre = "0.6.8"
 enum-map = "2.6.3"
 futures = "0.3.28"
@@ -211,7 +212,8 @@ tokio = { version = "1.32.0", features = [
     "parking_lot",
     "tracing",
 ] }
-tonic = "0.10.2"
+tonic = "0.11"
+tower = "0.4"
 tracing = "0.1.37"
 tracing-futures = { version = "0.2.5", features = ["futures-03"] }
 tracing-subscriber = "0.3"

deny.toml

@@ -26,29 +26,25 @@ all-features = true
 ignore = []
 
 [bans]
-multiple-versions = "deny"
 deny = [
     { crate = "openssl-sys", use-instead = "rustls" },
     { crate = "openssl", use-instead = "rustls" },
+    { crate = "cmake", use-instead = "cc" },
     { crate = "chrono", use-instead = "time", wrappers = [
         "k8s-openapi",
         "kube-client",
         "kube-core",
     ] },
 ]
+multiple-versions = "deny"
 skip = [
-    { crate = "[email protected]", reason = "multiple crates use this old version" },
-    { crate = "[email protected]", reason = "multiple crates use this old version" },
-    { crate = "[email protected]", reason = "multiple crates use this old version" },
-    { crate = "[email protected]", reason = "tower is the sole user of this old version" },
-    { crate = "[email protected]", reason = "used by the old version of indexmap" },
-    { crate = "[email protected]", reason = "used by lasso/libflate" },
-    { crate = "[email protected]", reason = "trust-dns uses an old version" },
-    { crate = "[email protected]", reason = "tokio-uring is the sole user of this old version" },
+    { crate = "[email protected]", reason = "several crates use this old version" },
 ]
 skip-tree = [
     { crate = "[email protected]", reason = "matchers is using an old version, https://github.com/hawkw/matchers/pull/5, but it's also barely maintained..." },
-    { crate = "[email protected]", reason = "We need the newer version of kube, but rustls crates in other deps aren't updated yet." },
+    { crate = "[email protected]", reason = "Uses _many_ outdated crates" },
+    # Much like trust-dns this pulls in a ton of outdated dependencies, but it's _slightly_ better
+    { crate = "[email protected]", reason = "Uses _many_ outdated crates" },
 ]
 
 # This section is considered when running `cargo deny check licenses`

src/components/admin.rs

@@ -16,12 +16,13 @@
 
 mod health;
 
-use std::convert::Infallible;
 use std::sync::Arc;
 use std::time::Duration;
 
-use hyper::service::{make_service_fn, service_fn};
-use hyper::{Body, Method, Request, Response, Server as HyperServer, StatusCode};
+use bytes::Bytes;
+use http_body_util::Full;
+use hyper::{Method, Request, Response, StatusCode};
+type Body = Full<Bytes>;
 
 use crate::config::Config;
 use health::Health;
@@ -55,7 +56,7 @@ impl Admin {
         &self,
         config: Arc<Config>,
         address: Option<std::net::SocketAddr>,
-    ) -> std::thread::JoinHandle<Result<(), hyper::Error>> {
+    ) -> std::thread::JoinHandle<eyre::Result<()>> {
         let address = address.unwrap_or_else(|| (std::net::Ipv6Addr::UNSPECIFIED, PORT).into());
         let health = Health::new();
         tracing::info!(address = %address, "Starting admin endpoint");
@@ -71,28 +72,43 @@ impl Admin {
                     .build()
                     .expect("couldn't create tokio runtime in thread");
                 runtime.block_on(async move {
-                    let make_svc = make_service_fn(move |_conn| {
-                        let config = config.clone();
-                        let health = health.clone();
-                        let mode = mode.clone();
-                        async move {
-                            let config = config.clone();
-                            let health = health.clone();
-                            let mode = mode.clone();
-                            Ok::<_, Infallible>(service_fn(move |req| {
+                    let accept_stream = tokio::net::TcpListener::bind(address).await?;
+                    let http_task: tokio::task::JoinHandle<eyre::Result<()>> =
+                        tokio::task::spawn(async move {
+                            loop {
+                                let (stream, _) = accept_stream.accept().await?;
+                                let stream = hyper_util::rt::TokioIo::new(stream);
+
                                 let config = config.clone();
                                 let health = health.clone();
                                 let mode = mode.clone();
-                                async move {
-                                    Ok::<_, Infallible>(
-                                        mode.handle_request(req, config, health).await,
-                                    )
-                                }
-                            }))
-                        }
-                    });
-
-                    HyperServer::bind(&address).serve(make_svc).await
+                                tokio::spawn(async move {
+                                    let svc = hyper::service::service_fn(move |req| {
+                                        let config = config.clone();
+                                        let health = health.clone();
+                                        let mode = mode.clone();
+
+                                        async move {
+                                            Ok::<_, std::convert::Infallible>(
+                                                mode.handle_request(req, config, health).await,
+                                            )
+                                        }
+                                    });
+
+                                    let svc = tower::ServiceBuilder::new().service(svc);
+                                    if let Err(err) = hyper::server::conn::http1::Builder::new()
+                                        .serve_connection(stream, svc)
+                                        .await
+                                    {
+                                        tracing::warn!(
+                                            "failed to reponse to phoenix request: {err}"
+                                        );
+                                    }
+                                });
+                            }
+                        });
+
+                    http_task.await?
                 })
             })
             .expect("failed to spawn admin-http thread")
@@ -111,7 +127,7 @@ impl Admin {
 
     async fn handle_request(
         &self,
-        request: Request<Body>,
+        request: Request<hyper::body::Incoming>,
         config: Arc<Config>,
         health: Health,
     ) -> Response<Body> {
@@ -133,7 +149,7 @@ impl Admin {
                         tracing::warn!(%error, "admin http server error");
                         Response::builder()
                             .status(StatusCode::INTERNAL_SERVER_ERROR)
-                            .body(Body::from("internal error"))
+                            .body(Body::new(Bytes::from("internal error")))
                             .unwrap()
                     }
                 }
@@ -146,15 +162,17 @@ impl Admin {
                         "Content-Type",
                         hyper::header::HeaderValue::from_static("application/json"),
                     )
-                    .body(Body::from(body))
+                    .body(Body::new(Bytes::from(body)))
                     .unwrap(),
                 Err(err) => Response::builder()
                     .status(StatusCode::INTERNAL_SERVER_ERROR)
-                    .body(Body::from(format!("failed to create config dump: {err}")))
+                    .body(Body::new(Bytes::from(format!(
+                        "failed to create config dump: {err}"
+                    ))))
                     .unwrap(),
             },
             (_, _) => {
-                let mut response = Response::new(Body::empty());
+                let mut response = Response::new(Body::new(Bytes::new()));
                 *response.status_mut() = StatusCode::NOT_FOUND;
                 response
             }
@@ -167,27 +185,27 @@ fn check_readiness(check: impl Fn() -> bool) -> Response<Body> {
         return Response::new("ok".into());
     }
 
-    let mut response = Response::new(Body::empty());
+    let mut response = Response::new(Body::new(Bytes::new()));
     *response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
     response
 }
 
 fn collect_metrics() -> Response<Body> {
-    let mut response = Response::new(Body::empty());
+    let mut response = Response::new(Body::new(Bytes::new()));
     let mut buffer = vec![];
     let encoder = prometheus::TextEncoder::new();
     let body =
         prometheus::Encoder::encode(&encoder, &crate::metrics::registry().gather(), &mut buffer)
             .map_err(|error| tracing::warn!(%error, "Failed to encode metrics"))
             .and_then(|_| {
                 String::from_utf8(buffer)
-                    .map(Body::from)
+                    .map(hyper::body::Bytes::from)
                     .map_err(|error| tracing::warn!(%error, "Failed to convert metrics to utf8"))
             });
 
     match body {
         Ok(body) => {
-            *response.body_mut() = body;
+            *response.body_mut() = Body::new(body);
         }
         Err(_) => {
             *response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
@@ -227,7 +245,7 @@ async fn collect_pprof(
         .header(hyper::header::CONTENT_LENGTH, gzip_body.len() as u64)
         .header(hyper::header::CONTENT_TYPE, "application/octet-stream")
         .header(hyper::header::CONTENT_ENCODING, "gzip")
-        .body(Body::from(gzip_body))
+        .body(Body::new(Bytes::from(gzip_body)))
         .map_err(From::from)
 }
 

src/components/admin/health.rs

@@ -16,7 +16,7 @@
 
 use std::sync::atomic::AtomicBool;
 
-use hyper::{Body, Response, StatusCode};
+use hyper::{Response, StatusCode};
 use std::panic;
 use std::sync::atomic::Ordering::Relaxed;
 use std::sync::Arc;
@@ -44,12 +44,12 @@ impl Health {
     }
 
     /// returns a HTTP 200 response if the proxy is healthy.
-    pub fn check_liveness(&self) -> Response<Body> {
+    pub fn check_liveness(&self) -> Response<http_body_util::Full<bytes::Bytes>> {
         if self.healthy.load(Relaxed) {
             return Response::new("ok".into());
         };
 
-        let mut response = Response::new(Body::empty());
+        let mut response = Response::new(http_body_util::Full::new(bytes::Bytes::new()));
         *response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
         response
     }

src/net/endpoint/address.rs

@@ -21,9 +21,9 @@ use std::{
     str::FromStr,
 };
 
+use hickory_resolver::TokioAsyncResolver;
 use once_cell::sync::Lazy;
 use serde::{Deserialize, Serialize};
-use trust_dns_resolver::{AsyncResolver, TokioAsyncResolver};
 
 use crate::generated::envoy::config::core::v3::{
     address::Address as EnvoyAddress, SocketAddress as EnvoySocketAddress,
@@ -62,7 +62,7 @@ impl EndpointAddress {
     /// if present.
     pub async fn to_socket_addr(&self) -> std::io::Result<SocketAddr> {
         static DNS: Lazy<TokioAsyncResolver> =
-            Lazy::new(|| AsyncResolver::tokio_from_system_conf().unwrap());
+            Lazy::new(|| TokioAsyncResolver::tokio_from_system_conf().unwrap());
 
         let ip = match &self.host {
             AddressKind::Ip(ip) => *ip,

src/net/maxmind_db.rs

@@ -1,18 +1,19 @@
 use std::sync::Arc;
 
 use bytes::Bytes;
+use hyper_util::client::legacy;
 use maxminddb::Reader;
 use once_cell::sync::Lazy;
 
 type Result<T, E = Error> = std::result::Result<T, E>;
 
 static HTTP: Lazy<
-    hyper::Client<
-        hyper_rustls::HttpsConnector<hyper::client::connect::HttpConnector>,
-        hyper::body::Body,
+    legacy::Client<
+        hyper_rustls::HttpsConnector<legacy::connect::HttpConnector>,
+        http_body_util::Empty<Bytes>,
     >,
 > = Lazy::new(|| {
-    hyper::Client::builder().build(
+    legacy::Client::builder(hyper_util::rt::TokioExecutor::new()).build(
         hyper_rustls::HttpsConnectorBuilder::new()
             .with_webpki_roots()
             .https_or_http()
@@ -107,12 +108,15 @@ impl MaxmindDb {
     #[tracing::instrument(skip_all, fields(url = %url))]
     pub async fn open_url(url: &url::Url) -> Result<Self> {
         tracing::info!("requesting maxmind database from network");
-        let data = hyper::body::to_bytes(
-            HTTP.get(url.as_str().try_into().unwrap())
-                .await?
-                .into_body(),
-        )
-        .await?;
+
+        use http_body_util::BodyExt;
+        let data = HTTP
+            .get(url.as_str().try_into().unwrap())
+            .await?
+            .into_body()
+            .collect()
+            .await?
+            .to_bytes();
 
         tracing::debug!("finished download");
         let reader = Reader::from_source(data)?;
@@ -183,6 +187,9 @@ pub enum Error {
     MaxmindDb(#[from] maxminddb::MaxMindDBError),
     #[error(transparent)]
     Http(#[from] hyper::Error),
+    #[error(transparent)]
+    HttpClient(#[from] legacy::Error),
+
     #[error(transparent)]
     Io(#[from] std::io::Error),
 }