chore: upgrade the xnet endpoint to hyper >1 (#1360)

Use the same tokio runtime for all http servers

I have ran `ict test
//rs/tests/message_routing/xnet:xnet_slo_120_subnets_staging_test`
multiple times and it always passed

---------

Co-authored-by: Stefan Neamtu <[email protected]>
Co-authored-by: IDX GitHub Automation <[email protected]>

rs/async_utils/src/lib.rs

@@ -48,7 +48,11 @@ pub async fn shutdown_signal(log: Logger) {
 /// Recommended way of starting a TCP listener given a socket addr. The function
 /// will panic if it cannot start the listener, because the OS error can't be
 /// handled by the caller.
-pub fn start_tcp_listener(local_addr: std::net::SocketAddr) -> tokio::net::TcpListener {
+pub fn start_tcp_listener(
+    local_addr: std::net::SocketAddr,
+    runtime_handle: &tokio::runtime::Handle,
+) -> tokio::net::TcpListener {
+    let _enter = runtime_handle.enter();
     let err_msg = format!("Could not start TCP listener at addr = {}", local_addr);
     let socket = if local_addr.is_ipv6() {
         tokio::net::TcpSocket::new_v6().expect(&err_msg)

rs/http_endpoints/metrics/src/lib.rs

@@ -149,11 +149,11 @@ impl MetricsHttpEndpoint {
     fn start_http(&self, address: SocketAddr) {
         // we need to enter the tokio context in order to create the timeout layer and the tcp
         // socket
-        let _enter = self.rt_handle.enter();
 
         let mut addr = "[::]:9090".parse::<SocketAddr>().unwrap();
         addr.set_port(address.port());
-        let tcp_listener = start_tcp_listener(addr);
+        let tcp_listener = start_tcp_listener(addr, &self.rt_handle);
+        let _enter: tokio::runtime::EnterGuard = self.rt_handle.enter();
         let metrics_service = get(metrics_endpoint)
             .layer(
                 ServiceBuilder::new()

rs/http_endpoints/public/src/lib.rs

@@ -304,14 +304,13 @@ pub fn start_server(
     let listen_addr = config.listen_addr;
     info!(log, "Starting HTTP server...");
 
-    let _enter = rt_handle.enter();
     // TODO(OR4-60): temporarily listen on [::] so that we accept both IPv4 and
     // IPv6 connections. This requires net.ipv6.bindv6only = 0. Revert this once
     // we have rolled out IPv6 in prometheus and ic_p8s_service_discovery.
     let mut addr = "[::]:8080".parse::<SocketAddr>().unwrap();
     addr.set_port(listen_addr.port());
-    let tcp_listener = start_tcp_listener(addr);
-
+    let tcp_listener = start_tcp_listener(addr, &rt_handle);
+    let _enter = rt_handle.enter();
     if !AtomicCell::<ReplicaHealthStatus>::is_lock_free() {
         error!(log, "Replica health status uses locks instead of atomics.");
     }

rs/replica/src/setup_ic_stack.rs

@@ -254,7 +254,7 @@ pub fn construct_ic_stack(
     let message_router = Arc::new(message_router);
     let xnet_config = XNetEndpointConfig::from(Arc::clone(&registry) as Arc<_>, node_id, log);
     let xnet_endpoint = XNetEndpoint::new(
-        rt_handle_xnet.clone(),
+        rt_handle_http.clone(),
         Arc::clone(&certified_stream_store),
         Arc::clone(&crypto) as Arc<_>,
         registry.clone(),

rs/xnet/endpoint/BUILD.bazel

@@ -4,6 +4,7 @@ package(default_visibility = ["//visibility:public"])
 
 DEPENDENCIES = [
     # Keep sorted.
+    "//rs/async_utils",
     "//rs/crypto/tls_interfaces",
     "//rs/interfaces/certified_stream_store",
     "//rs/interfaces/registry",
@@ -12,13 +13,17 @@ DEPENDENCIES = [
     "//rs/protobuf",
     "//rs/registry/helpers",
     "//rs/types/types",
-    "//rs/xnet/hyper",
-    "@crate_index//:hyper_0_14_27",
+    "@crate_index//:axum",
+    "@crate_index//:crossbeam-channel",
+    "@crate_index//:hyper",
+    "@crate_index//:hyper-util",
     "@crate_index//:prometheus",
     "@crate_index//:serde",
     "@crate_index//:serde_json",
     "@crate_index//:slog",
     "@crate_index//:tokio",
+    "@crate_index//:tokio-rustls",
+    "@crate_index//:tower",
     "@crate_index//:url",
 ]
 

rs/xnet/endpoint/Cargo.toml

@@ -7,7 +7,11 @@ description.workspace = true
 documentation.workspace = true
 
 [dependencies]
-hyper = { version = "0.14.18", features = ["full", "tcp"] }
+axum = { workspace = true }
+hyper = { workspace = true }
+hyper-util = { workspace = true }
+crossbeam-channel = { workspace = true }
+ic-async-utils = { path = "../../async_utils" }
 ic-crypto-tls-interfaces = { path = "../../crypto/tls_interfaces" }
 ic-interfaces-certified-stream-store = { path = "../../interfaces/certified_stream_store" }
 ic-interfaces-registry = { path = "../../interfaces/registry" }
@@ -16,12 +20,13 @@ ic-metrics = { path = "../../monitoring/metrics" }
 ic-protobuf = { path = "../../protobuf" }
 ic-registry-client-helpers = { path = "../../registry/helpers" }
 ic-types = { path = "../../types/types" }
-ic-xnet-hyper = { path = "../hyper" }
 prometheus = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 slog = { workspace = true }
 tokio = { workspace = true }
+tokio-rustls = { workspace = true }
+tower = { workspace = true }
 url = { workspace = true }
 
 [dev-dependencies]