Create axum_utils and begin VC HTTP API migration

Cargo.toml

@@ -18,6 +18,7 @@ members = [
     "beacon_node/timer",
     "boot_node",
     "common/account_utils",
+    "common/axum_utils",
     "common/clap_utils",
     "common/compare_fields",
     "common/compare_fields_derive",
@@ -105,7 +106,8 @@ alloy-rlp = "0.3.4"
 anyhow = "1"
 arbitrary = { version = "1", features = ["derive"] }
 async-channel = "1.9.0"
-axum = "0.7.7"
+axum = "0.8"
+axum_utils = { path = "common/axum_utils" }
 beacon_chain = { path = "beacon_node/beacon_chain" }
 beacon_node = { path = "beacon_node" }
 beacon_node_fallback = { path = "validator_client/beacon_node_fallback" }
@@ -250,6 +252,7 @@ sysinfo = "0.26"
 system_health = { path = "common/system_health" }
 task_executor = { path = "common/task_executor" }
 tempfile = "3"
+thiserror = "2"
 tokio = { version = "1", features = [
     "rt-multi-thread",
     "sync",
@@ -258,6 +261,7 @@ tokio = { version = "1", features = [
 ] }
 tokio-stream = { version = "0.1", features = ["sync"] }
 tokio-util = { version = "0.7", features = ["codec", "compat", "time"] }
+tower = "0.5"
 tracing = "0.1.40"
 tracing-appender = "0.2"
 tracing-core = "0.1"
@@ -279,6 +283,7 @@ validator_store = { path = "validator_client/validator_store" }
 validator_test_rig = { path = "testing/validator_test_rig" }
 warp = { version = "0.3.7", default-features = false, features = ["tls"] }
 warp_utils = { path = "common/warp_utils" }
+warpdrive = "0.1"
 workspace_members = { path = "common/workspace_members" }
 xdelta3 = { git = "https://github.com/sigp/xdelta3-rs", rev = "4db64086bb02e9febb584ba93b9d16bb2ae3825a" }
 zeroize = { version = "1", features = ["zeroize_derive", "serde"] }

beacon_node/Cargo.toml

@@ -19,6 +19,7 @@ testing = [] # Enables testing-only CLI flags
 
 [dependencies]
 account_utils = { workspace = true }
+axum_utils = { workspace = true }
 beacon_chain = { workspace = true }
 clap = { workspace = true }
 clap_utils = { workspace = true }

beacon_node/client/src/builder.rs

@@ -597,7 +597,7 @@ where
     ///
     /// If type inference errors are being raised, see the comment on the definition of `Self`.
     #[allow(clippy::type_complexity)]
-    pub fn build(
+    pub async fn build(
         mut self,
     ) -> Result<Client<Witness<TSlotClock, E, THotStore, TColdStore>>, String> {
         let runtime_context = self
@@ -656,6 +656,7 @@ where
             let exit = runtime_context.executor.exit();
 
             let (listen_addr, server) = http_metrics::serve(ctx, exit)
+                .await
                 .map_err(|e| format!("Unable to start HTTP metrics server: {:?}", e))?;
 
             runtime_context

beacon_node/http_api/Cargo.toml

@@ -6,6 +6,7 @@ edition = { workspace = true }
 autotests = false                               # using a single test binary compiles faster
 
 [dependencies]
+axum_utils = { workspace = true }
 beacon_chain = { workspace = true }
 beacon_processor = { workspace = true }
 bs58 = "0.4.0"

beacon_node/http_api/src/lib.rs

@@ -35,6 +35,7 @@ mod version;
 use crate::light_client::{get_light_client_bootstrap, get_light_client_updates};
 use crate::produce_block::{produce_blinded_block_v2, produce_block_v2, produce_block_v3};
 use crate::version::beacon_response;
+use axum_utils::tls::TlsConfig;
 use beacon_chain::{
     AttestationError as AttnError, BeaconChain, BeaconChainError, BeaconChainTypes,
     WhenSlotSkipped, attestation_verification::VerifiedAttestation,
@@ -116,13 +117,6 @@ type HttpServer = (SocketAddr, Pin<Box<dyn Future<Output = ()> + Send>>);
 /// Alias for readability.
 pub type ExecutionOptimistic = bool;
 
-/// Configuration used when serving the HTTP server over TLS.
-#[derive(PartialEq, Debug, Clone, Serialize, Deserialize)]
-pub struct TlsConfig {
-    pub cert: PathBuf,
-    pub key: PathBuf,
-}
-
 /// A wrapper around all the items required to spawn the HTTP server.
 ///
 /// The server will gracefully handle the case where any fields are `None`.

beacon_node/http_metrics/Cargo.toml

@@ -6,10 +6,11 @@ edition = { workspace = true }
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+axum = { workspace = true }
+axum_utils = { workspace = true }
 beacon_chain = { workspace = true }
 health_metrics = { workspace = true }
 lighthouse_network = { workspace = true }
-lighthouse_version = { workspace = true }
 logging = { workspace = true }
 malloc_utils = { workspace = true }
 metrics = { workspace = true }
@@ -18,8 +19,6 @@ serde = { workspace = true }
 slot_clock = { workspace = true }
 store = { workspace = true }
 tracing = { workspace = true }
-warp = { workspace = true }
-warp_utils = { workspace = true }
 
 [dev-dependencies]
 logging = { workspace = true }

beacon_node/http_metrics/src/lib.rs

@@ -3,27 +3,40 @@
 //! For other endpoints, see the `http_api` crate.
 mod metrics;
 
+use axum::{
+    Router,
+    extract::State,
+    http::{StatusCode, header},
+    middleware,
+    response::IntoResponse,
+    routing::get,
+};
+use axum_utils::{Server, cors::build_cors_layer, middleware::add_server_header};
 use beacon_chain::{BeaconChain, BeaconChainTypes};
 use lighthouse_network::prometheus_client::registry::Registry;
-use lighthouse_version::version_with_platform;
 use logging::crit;
 use serde::{Deserialize, Serialize};
 use std::future::Future;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::path::PathBuf;
 use std::sync::Arc;
 use tracing::info;
-use warp::{Filter, http::Response};
 
 #[derive(Debug)]
 pub enum Error {
-    Warp(warp::Error),
+    ServerError(String),
     Other(String),
 }
 
-impl From<warp::Error> for Error {
-    fn from(e: warp::Error) -> Self {
-        Error::Warp(e)
+impl From<axum_utils::server::BuilderError> for Error {
+    fn from(e: axum_utils::server::BuilderError) -> Self {
+        Error::ServerError(e.to_string())
+    }
+}
+
+impl From<axum_utils::cors::CorsError> for Error {
+    fn from(e: axum_utils::cors::CorsError) -> Self {
+        Error::ServerError(e.to_string())
     }
 }
 
@@ -81,25 +94,12 @@ impl Default for Config {
 ///
 /// Returns an error if the server is unable to bind or there is another error during
 /// configuration.
-pub fn serve<T: BeaconChainTypes>(
+pub async fn serve<T: BeaconChainTypes>(
     ctx: Arc<Context<T>>,
     shutdown: impl Future<Output = ()> + Send + Sync + 'static,
 ) -> Result<(SocketAddr, impl Future<Output = ()>), Error> {
     let config = &ctx.config;
 
-    // Configure CORS.
-    let cors_builder = {
-        let builder = warp::cors()
-            .allow_method("GET")
-            .allow_headers(vec!["Content-Type"]);
-
-        warp_utils::cors::set_builder_origins(
-            builder,
-            config.allow_origin.as_deref(),
-            (config.listen_addr, config.listen_port),
-        )?
-    };
-
     // Sanity check.
     if !config.enabled {
         crit!("Cannot start disabled metrics HTTP server");
@@ -108,44 +108,53 @@ pub fn serve<T: BeaconChainTypes>(
         ));
     }
 
-    let inner_ctx = ctx.clone();
-    let routes = warp::get()
-        .and(warp::path("metrics"))
-        .map(move || inner_ctx.clone())
-        .and_then(|ctx: Arc<Context<T>>| async move {
-            Ok::<_, warp::Rejection>(
-                metrics::gather_prometheus_metrics(&ctx)
-                    .map(|body| {
-                        Response::builder()
-                            .status(200)
-                            .header("Content-Type", "text/plain")
-                            .body(body)
-                            .unwrap()
-                    })
-                    .unwrap_or_else(|e| {
-                        Response::builder()
-                            .status(500)
-                            .header("Content-Type", "text/plain")
-                            .body(format!("Unable to gather metrics: {:?}", e))
-                            .unwrap()
-                    }),
-            )
-        })
-        // Add a `Server` header.
-        .map(|reply| warp::reply::with_header(reply, "Server", &version_with_platform()))
-        .with(cors_builder.build());
-
-    let (listening_socket, server) = warp::serve(routes).try_bind_with_graceful_shutdown(
-        SocketAddr::new(config.listen_addr, config.listen_port),
-        async {
-            shutdown.await;
-        },
+    let cors_layer = build_cors_layer(
+        config.allow_origin.as_deref(),
+        config.listen_addr,
+        config.listen_port,
     )?;
 
+    let router = Router::new()
+        .route("/metrics", get(metrics_handler::<T>))
+        .with_state(ctx.clone())
+        .layer(cors_layer)
+        .layer(middleware::from_fn(add_server_header));
+
+    let address = SocketAddr::new(config.listen_addr, config.listen_port);
+    let server = Server::builder()
+        .router(router)
+        .address(address)
+        .build()
+        .await?;
+
+    let (address, server) = server
+        .serve_with_shutdown(shutdown)
+        .await
+        .map_err(|e| Error::ServerError(e.to_string()))?;
+
     info!(
-        listen_address = listening_socket.to_string(),
+        listen_address = %address,
         "Metrics HTTP server started"
     );
 
-    Ok((listening_socket, server))
+    let server_future = async move {
+        if let Err(e) = server.await {
+            tracing::error!(error = ?e, "Metrics HTTP server error");
+        }
+    };
+
+    Ok((address, server_future))
+}
+
+async fn metrics_handler<T: BeaconChainTypes>(
+    State(ctx): State<Arc<Context<T>>>,
+) -> impl IntoResponse {
+    match metrics::gather_prometheus_metrics(&ctx) {
+        Ok(body) => (StatusCode::OK, [(header::CONTENT_TYPE, "text/plain")], body),
+        Err(e) => (
+            StatusCode::INTERNAL_SERVER_ERROR,
+            [(header::CONTENT_TYPE, "text/plain")],
+            format!("Unable to gather metrics: {:?}", e),
+        ),
+    }
 }

beacon_node/http_metrics/tests/tests.rs

@@ -34,7 +34,7 @@ async fn returns_200_ok() {
             // It's not really interesting why this triggered, just that it happened.
             let _ = shutdown_rx.await;
         };
-        let (listening_socket, server) = http_metrics::serve(ctx, server_shutdown).unwrap();
+        let (listening_socket, server) = http_metrics::serve(ctx, server_shutdown).await.unwrap();
 
         tokio::spawn(server);
 

beacon_node/src/config.rs

@@ -1,4 +1,5 @@
 use account_utils::{STDIN_INPUTS_FLAG, read_input_from_user};
+use axum_utils::tls::TlsConfig;
 use beacon_chain::chain_config::{
     DEFAULT_PREPARE_PAYLOAD_LOOKAHEAD_FACTOR, DEFAULT_RE_ORG_HEAD_THRESHOLD,
     DEFAULT_RE_ORG_MAX_EPOCHS_SINCE_FINALIZATION, DEFAULT_RE_ORG_PARENT_THRESHOLD,
@@ -12,7 +13,6 @@ use client::{ClientConfig, ClientGenesis};
 use directory::{DEFAULT_BEACON_NODE_DIR, DEFAULT_NETWORK_DIR, DEFAULT_ROOT_DIR};
 use environment::RuntimeContext;
 use execution_layer::DEFAULT_JWT_FILE;
-use http_api::TlsConfig;
 use lighthouse_network::{Enr, Multiaddr, NetworkConfig, PeerIdSerialized, multiaddr::Protocol};
 use network_utils::listen_addr::ListenAddress;
 use sensitive_url::SensitiveUrl;

beacon_node/src/lib.rs

@@ -138,6 +138,7 @@ impl<E: EthSpec> ProductionBeaconNode<E> {
             .notifier()?
             .http_metrics_config(client_config.http_metrics.clone())
             .build()
+            .await
             .map(Self)
     }
 

common/axum_utils/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "axum_utils"
+version = "0.1.0"
+authors = ["Sigma Prime <[email protected]>"]
+edition = { workspace = true }
+
+[dependencies]
+axum = { workspace = true }
+axum-server = { version = "0.7", features = ["tls-rustls-no-provider"] }
+http = "1"
+lighthouse_version = { workspace = true }
+rustls = { version = "0.23", default-features = false, features = ["ring"] }
+serde = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+tower-http = { version = "0.6", features = ["cors"] }