Double HBONE implementation for ambient multicluster

src/proxy/outbound.rs

@@ -219,40 +219,56 @@ impl OutboundConnection {
         req: &Request,
         connection_stats: &ConnectionResult,
     ) -> Result<(), Error> {
-        // Outer HBONE
-        let upgraded = Box::pin(self.send_hbone_request(remote_addr, req)).await?;
-
-        // Inner HBONE
-        let upgraded = TokioH2Stream::new(upgraded);
+        // TODO: dst should take a hostname? and upstream_sans currently contains E/W Gateway certs
         let inner_workload = pool::WorkloadKey {
             src_id: req.source.identity(),
             dst_id: req.upstream_sans.clone(),
             src: remote_addr.ip(),
             dst: req.actual_destination,
         };
         let request = self.create_hbone_request(remote_addr, req);
+        let workload_key = &inner_workload;
 
-        let (_conn_client, inner_upgraded, drain_tx, driver_task) = self
-            .pool
-            .send_request_unpooled(upgraded, &inner_workload, request)
+        // To shut down inner HTTP2 connection
+        let (drain_tx, drain_rx) = tokio::sync::watch::channel(false);
+        let key = workload_key.clone();
+        let dest = rustls::pki_types::ServerName::IpAddress(key.dst.ip().into());
+        let cert = self
+            .pi
+            .local_workload_information
+            .fetch_certificate()
             .await?;
-        let inner_upgraded = inner_upgraded?;
+        // FIXME The following isn't great because it will also contain the identity of the E/W gateways
+        let connector = cert.outbound_connector(req.upstream_sans.clone())?;
+
+        // Do actual IO as late as possible
+        // Outer HBONE
+        let upgraded = Box::pin(self.send_hbone_request(remote_addr, req)).await?;
+        // Wrap upgraded to implement tokio's Async{Write,Read}
+        let upgraded = TokioH2Stream::new(upgraded);
+        let tls_stream = connector.connect(upgraded, dest).await?;
+
+        let (sender, driver_task) =
+            super::h2::client::spawn_connection(self.pi.cfg.clone(), tls_stream, drain_rx).await?;
+        let mut connection = super::pool::ConnClient {
+            sender,
+            wl_key: key,
+        };
+        let inner_upgraded = connection.sender.send_request(request).await?;
         let res = copy::copy_bidirectional(
             copy::TcpStreamSplitter(stream),
             inner_upgraded,
             connection_stats,
         )
         .await;
 
-        // This always drops ungracefully
-        // drop(conn_client);
-        // tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-        // drain_tx.send(true).unwrap();
-        // tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-        drain_tx.send(true).unwrap();
+        let _ = drain_tx.send(true);
         let _ = driver_task.await;
-        // this sleep is important, so we have a race condition somewhere
-        // tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+        // Here, there is an implicit, drop(conn_client).
+        // Its really important that this happens AFTER driver_task finishes.
+        // Otherwise, TLS connections do not terminate gracefully.
+        //
+        // drop(conn_client);
         res
     }
 

src/proxy/pool.rs

@@ -71,7 +71,7 @@ struct PoolState {
     // This is merely a counter to track the overall number of conns this pool spawns
     // to ensure we get unique poolkeys-per-new-conn, it is not a limit
     pool_global_conn_count: AtomicI32,
-    pub spawner: ConnSpawner,
+    spawner: ConnSpawner,
 }
 
 struct ConnSpawner {
@@ -413,17 +413,28 @@ impl WorkloadHBONEPool {
         Error,
     > {
         let (tx, rx) = tokio::sync::watch::channel(false);
-        let (mut connection, driver_task) = self
+        let key = workload_key.clone();
+        let dest = rustls::pki_types::ServerName::IpAddress(key.dst.ip().into());
+        let cert = self
             .state
             .spawner
-            .new_unpooled_conn(workload_key.clone(), stream, rx)
+            .local_workload
+            .fetch_certificate()
             .await?;
+        let connector = cert.outbound_connector(vec![])?;
+        let tls_stream = connector.connect(stream, dest).await?;
+        let (sender, driver_drain) =
+            h2::client::spawn_connection(self.state.spawner.cfg.clone(), tls_stream, rx).await?;
+        let mut connection = ConnClient {
+            sender,
+            wl_key: key,
+        };
 
         Ok((
             connection.clone(),
             connection.sender.send_request(request).await,
             tx,
-            driver_task,
+            driver_drain,
         ))
     }
 
@@ -568,9 +579,9 @@ impl WorkloadHBONEPool {
 // A sort of faux-client, that represents a single checked-out 'request sender' which might
 // send requests over some underlying stream using some underlying http/2 client
 pub struct ConnClient {
-    sender: H2ConnectClient,
+    pub sender: H2ConnectClient,
     // 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.
+    pub wl_key: WorkloadKey, // the WL key associated with this client.
 }
 
 impl ConnClient {