Format

Signed-off-by: Benjamin Leggett <benjamin.leggett@solo.io>

src/admin.rs

@@ -395,7 +395,11 @@ fn change_log_level(reset: bool, level: &str) -> Response<Full<Bytes>> {
 async fn handle_jemalloc_pprof_heapgen(
     _req: Request<Incoming>,
 ) -> anyhow::Result<Response<Full<Bytes>>> {
-    let mut prof_ctl = jemalloc_pprof::PROF_CTL.as_ref().expect("should init").lock().await;
+    let mut prof_ctl = jemalloc_pprof::PROF_CTL
+        .as_ref()
+        .expect("should init")
+        .lock()
+        .await;
     if !prof_ctl.activated() {
         return Ok(Response::builder()
             .status(hyper::StatusCode::INTERNAL_SERVER_ERROR)

src/proxy/outbound.rs

@@ -80,9 +80,10 @@ impl Outbound {
         let (sub_drain_signal, sub_drain) = drain::channel();
 
         let pool = proxy::pool::WorkloadHBONEPool::new(
-                    self.pi.cfg.clone(),
-                    self.pi.socket_factory.clone(),
-                    self.pi.cert_manager.clone());
+            self.pi.cfg.clone(),
+            self.pi.socket_factory.clone(),
+            self.pi.cert_manager.clone(),
+        );
         let accept = async move {
             loop {
                 // Asynchronously wait for an inbound socket.
@@ -147,13 +148,8 @@ impl OutboundConnection {
         let source_addr =
             socket::to_canonical(source_stream.peer_addr().expect("must receive peer addr"));
         let dst_addr = socket::orig_dst_addr_or_default(&source_stream);
-        self.proxy_to(
-            source_stream,
-            source_addr,
-            dst_addr,
-            false,
-        )
-        .await;
+        self.proxy_to(source_stream, source_addr, dst_addr, false)
+            .await;
     }
 
     // this is a cancellable outbound proxy. If `out_drain` is a Watch drain, will resolve
@@ -251,13 +247,8 @@ impl OutboundConnection {
 
         let res = match req.protocol {
             Protocol::HBONE => {
-                self.proxy_to_hbone(
-                    &mut source_stream,
-                    source_addr,
-                    &req,
-                    &result_tracker,
-                )
-                .await
+                self.proxy_to_hbone(&mut source_stream, source_addr, &req, &result_tracker)
+                    .await
             }
             Protocol::TCP => {
                 self.proxy_to_tcp(&mut source_stream, &req, &result_tracker)
@@ -304,9 +295,11 @@ impl OutboundConnection {
         };
 
         debug!("outbound - connection get START");
-        let mut connection = self.pool.connect(pool_key.clone())
-                                         .instrument(trace_span!("get pool conn"))
-                                         .await?;
+        let mut connection = self
+            .pool
+            .connect(pool_key.clone())
+            .instrument(trace_span!("get pool conn"))
+            .await?;
         debug!("outbound - connection get END");
 
         let mut f = http_types::proxies::Forwarded::new();
@@ -329,9 +322,10 @@ impl OutboundConnection {
         // There are scenarios (upstream hangup, etc) where this "send" will simply get stuck.
         // As in, stream processing deadlocks, and `send_request` never resolves to anything.
         // Probably related to https://github.com/hyperium/hyper/issues/3623
-        let response = connection.send_request(request)
-                                 .instrument(trace_span!("send pool conn"))
-                                 .await?;
+        let response = connection
+            .send_request(request)
+            .instrument(trace_span!("send pool conn"))
+            .await?;
         debug!("outbound - connection send END");
 
         let code = response.status();
@@ -669,10 +663,7 @@ mod tests {
                 connection_manager: ConnectionManager::default(),
             },
             id: TraceParent::new(),
-            pool: pool::WorkloadHBONEPool::new(
-                cfg,
-                sock_fact,
-                cert_mgr.clone()),
+            pool: pool::WorkloadHBONEPool::new(cfg, sock_fact, cert_mgr.clone()),
         };
 
         let req = outbound

src/proxy/pool.rs

@@ -14,11 +14,11 @@
 
 use super::{Error, SocketFactory};
 use bytes::Bytes;
-use std::time::Duration;
 use http_body_util::Empty;
 use hyper::body::Incoming;
 use hyper::client::conn::http2;
 use hyper::http::{Request, Response};
+use std::time::Duration;
 
 use std::collections::hash_map::DefaultHasher;
 use std::future::Future;
@@ -103,8 +103,13 @@ impl PoolState {
         let pool_key_ref = pool_key.clone();
         let release_timeout = self.pool_unused_release_timeout;
         self.close_pollers.push(tokio::spawn(async move {
-            pool_ref.idle_timeout(&pool_key_ref, release_timeout, evict, rx, pickup).await;
-            debug!("connection {:#?} was removed/checked out/timed out of the pool", pool_key_ref)
+            pool_ref
+                .idle_timeout(&pool_key_ref, release_timeout, evict, rx, pickup)
+                .await;
+            debug!(
+                "connection {:#?} was removed/checked out/timed out of the pool",
+                pool_key_ref
+            )
         }));
         let _ = self.pool_notifier.send(true);
     }
@@ -121,11 +126,17 @@ impl PoolState {
                 debug!("first checkout INNER WRITELOCK");
                 let _conn_lock = exist_conn_lock.as_ref().unwrap().lock().await;
 
-                debug!("getting conn for key {:#?} and hash {:#?}", workload_key, pool_key.key);
+                debug!(
+                    "getting conn for key {:#?} and hash {:#?}",
+                    workload_key, pool_key.key
+                );
                 self.connected_pool.get(&pool_key.key).and_then(|e_conn| {
                     debug!("got existing conn for key {:#?}", workload_key);
                     if e_conn.at_max_streamcount() {
-                        debug!("got conn for key {:#?}, but streamcount is maxed", workload_key);
+                        debug!(
+                            "got conn for key {:#?}, but streamcount is maxed",
+                            workload_key
+                        );
                         None
                     } else {
                         self.checkin_conn(e_conn.clone(), pool_key.clone());
@@ -136,7 +147,6 @@ impl PoolState {
             None => None,
         }
     }
-
 }
 
 impl Drop for PoolState {
@@ -159,12 +169,9 @@ impl WorkloadHBONEPool {
         let (timeout_send, timeout_recv) = watch::channel(false);
         let max_count = cfg.pool_max_streams_per_conn;
         let pool_duration = cfg.pool_unused_release_timeout;
-        debug!(
-            "constructing pool with {:#?} streams per conn",
-            max_count
-        );
+        debug!("constructing pool with {:#?} streams per conn", max_count);
         Self {
-            state: Arc::new(PoolState{
+            state: Arc::new(PoolState {
                 pool_notifier: timeout_tx,
                 timeout_tx: timeout_send,
                 timeout_rx: timeout_recv,
@@ -211,7 +218,10 @@ impl WorkloadHBONEPool {
         //
         // This is so we can backpressure correctly if 1000 tasks all demand a new connection
         // to the same key at once, and not eagerly open 1000 tunnel connections.
-        let existing_conn = self.state.first_checkout_conn_from_pool(&workload_key, &pool_key).await;
+        let existing_conn = self
+            .state
+            .first_checkout_conn_from_pool(&workload_key, &pool_key)
+            .await;
 
         debug!("pool connect GOT EXISTING");
         if existing_conn.is_some() {
@@ -406,7 +416,6 @@ impl WorkloadHBONEPool {
                 }
             }
         })
-
     }
 
     async fn spawn_new_pool_conn(
@@ -461,7 +470,6 @@ impl WorkloadHBONEPool {
 
         Ok(request_sender)
     }
-
 }
 
 #[derive(Debug, Clone)]
@@ -470,7 +478,7 @@ impl WorkloadHBONEPool {
 pub struct ConnClient {
     sender: http2::SendRequest<Empty<Bytes>>,
     stream_count: Arc<AtomicU16>, // the current streamcount for this client conn.
-    stream_count_max: u16, // the max streamcount associated with this client.
+    stream_count_max: u16,        // the max streamcount associated with this client.
     // A WL key may have many clients, but every client has no more than one WL key
     wl_key: WorkloadKey, // the WL key associated with this client.
 }
@@ -496,12 +504,13 @@ impl ConnClient {
 
     pub fn is_for_workload(&self, wl_key: WorkloadKey) -> Result<(), crate::proxy::Error> {
         if !(self.wl_key == wl_key) {
-            Err(crate::proxy::Error::Generic("fetched connection does not match workload key!".into()))
+            Err(crate::proxy::Error::Generic(
+                "fetched connection does not match workload key!".into(),
+            ))
         } else {
             Ok(())
         }
     }
-
 }
 
 #[derive(PartialEq, Eq, Hash, Clone, Debug)]

src/proxy/socks5.rs

@@ -68,9 +68,10 @@ impl Socks5 {
                 // TODO creating a new HBONE pool for SOCKS5 here may not be ideal,
                 // but ProxyInfo is overloaded and only `outbound` should ever use the pool.
                 let pool = crate::proxy::pool::WorkloadHBONEPool::new(
-                            self.pi.cfg.clone(),
-                            self.pi.socket_factory.clone(),
-                            self.pi.cert_manager.clone());
+                    self.pi.cfg.clone(),
+                    self.pi.socket_factory.clone(),
+                    self.pi.cert_manager.clone(),
+                );
                 match socket {
                     Ok((stream, remote)) => {
                         info!("accepted outbound connection from {}", remote);