proxy: add HTTP/1.1 Upgrade support automatically (linkerd#1126)

Any HTTP/1.1 requests seen by the proxy will automatically set up to prepare such that if the proxied responses agree to an upgrade, the two connections will converted into a standard TCP proxy duplex. Implementation ----------------- This adds a new type, `transparency::Http11Upgrade`, which is a sort of rendezvous type for triggering HTTP/1.1 upgrades. In the h1 server service, if a request looks like an upgrade (`h1::wants_upgrade`), the request body is decorated with this new `Http11Upgrade` type. It is actually a pair, and so the second half is put into the request extensions, so that the h1 client service may look for it right before serialization. If it finds the half in the extensions, it decorates the *response* body with that half (if it looks like a response upgrade (`h1::is_upgrade`)). The `HttpBody` type now has a `Drop` impl, which will look to see if its been decorated with an `Http11Upgrade` half. If so, it will check for hyper's new `Body::on_upgrade()` future, and insert that into the half. When both `Http11Upgrade` halves are dropped, its internal `Drop` will look to if both halves have supplied an upgrade. If so, the two `OnUpgrade` futures from hyper are joined on, and when they succeed, a `transparency::tcp::duplex()` future is created. This chain is spawned into the default executor. The `drain::Watch` signal is carried along, to ensure upgraded connections still count towards active connections when the proxy wants to shutdown. Closes linkerd#195
sprt · Jun 20, 2018 · c52385d · c52385d
1 parent 8dcb95d
commit c52385d
Show file tree

Hide file tree

Showing 13 changed files with 538 additions and 127 deletions.
diff --git a/Cargo.lock b/Cargo.lock
@@ -134,7 +134,7 @@ dependencies = [
  "h2 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)",
  "http 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
  "httparse 1.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
- "hyper 0.12.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "hyper 0.12.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "indexmap 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "inotify 0.5.2-dev (git+https://github.com/inotify-rs/inotify)",
  "ipnet 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -165,6 +165,7 @@ dependencies = [
  "tower-service 0.1.0 (git+https://github.com/tower-rs/tower)",
  "tower-util 0.1.0 (git+https://github.com/tower-rs/tower)",
  "trust-dns-resolver 0.9.0 (git+https://github.com/bluejekyll/trust-dns)",
+ "try-lock 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "untrusted 0.6.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "webpki 0.18.0-alpha4 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -416,7 +417,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 
 [[package]]
 name = "hyper"
-version = "0.12.1"
+version = "0.12.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "bytes 0.4.7 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -435,7 +436,7 @@ dependencies = [
  "tokio-reactor 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-tcp 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-timer 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
- "want 0.0.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "want 0.0.5 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -1328,7 +1329,7 @@ dependencies = [
 
 [[package]]
 name = "try-lock"
-version = "0.1.0"
+version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
 [[package]]
@@ -1399,12 +1400,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 
 [[package]]
 name = "want"
-version = "0.0.4"
+version = "0.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "try-lock 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "try-lock 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -1534,7 +1535,7 @@ dependencies = [
 "checksum hostname 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)" = "58fab6e177434b0bb4cd344a4dabaa5bd6d7a8d792b1885aebcae7af1091d1cb"
 "checksum http 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "75df369fd52c60635208a4d3e694777c099569b3dcf4844df8f652dc004644ab"
 "checksum httparse 1.2.4 (registry+https://github.com/rust-lang/crates.io-index)" = "c2f407128745b78abc95c0ffbe4e5d37427fdc0d45470710cfef8c44522a2e37"
-"checksum hyper 0.12.1 (registry+https://github.com/rust-lang/crates.io-index)" = "6416251e6672bff06fe96a3337570772845a44500fba2d178e2e55e0fab58a86"
+"checksum hyper 0.12.2 (registry+https://github.com/rust-lang/crates.io-index)" = "ad39a4f15051ccd4ea6adf44df851e00fd9062c71734391d806246b94e69dc1f"
 "checksum idna 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)" = "014b298351066f1512874135335d62a789ffe78a9974f94b43ed5621951eaf7d"
 "checksum indexmap 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "7b9378f1f3923647a9aea6af4c6b5de68cc8a71415459ad25ef191191c48f5b7"
 "checksum inotify 0.5.2-dev (git+https://github.com/inotify-rs/inotify)" = "<none>"
@@ -1631,7 +1632,7 @@ dependencies = [
 "checksum tower-util 0.1.0 (git+https://github.com/tower-rs/tower)" = "<none>"
 "checksum trust-dns-proto 0.4.0 (git+https://github.com/bluejekyll/trust-dns)" = "<none>"
 "checksum trust-dns-resolver 0.9.0 (git+https://github.com/bluejekyll/trust-dns)" = "<none>"
-"checksum try-lock 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "ee2aa4715743892880f70885373966c83d73ef1b0838a664ef0c76fffd35e7c2"
+"checksum try-lock 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "119b532a17fbe772d360be65617310164549a07c25a1deab04c84168ce0d4545"
 "checksum ucd-util 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "fd2be2d6639d0f8fe6cdda291ad456e23629558d466e2789d2c3e9892bda285d"
 "checksum unicode-bidi 0.3.4 (registry+https://github.com/rust-lang/crates.io-index)" = "49f2bd0c6468a8230e1db229cff8029217cf623c767ea5d60bfbd42729ea54d5"
 "checksum unicode-normalization 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "51ccda9ef9efa3f7ef5d91e8f9b83bbe6955f9bf86aec89d5cce2c874625920f"
@@ -1643,7 +1644,7 @@ dependencies = [
 "checksum url 1.7.0 (registry+https://github.com/rust-lang/crates.io-index)" = "f808aadd8cfec6ef90e4a14eb46f24511824d1ac596b9682703c87056c8678b7"
 "checksum utf8-ranges 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "662fab6525a98beff2921d7f61a39e7d59e0b425ebc7d0d9e66d316e55124122"
 "checksum void 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "6a02e4885ed3bc0f2de90ea6dd45ebcbb66dacffe03547fadbb0eeae2770887d"
-"checksum want 0.0.4 (registry+https://github.com/rust-lang/crates.io-index)" = "a05d9d966753fa4b5c8db73fcab5eed4549cfe0e1e4e66911e5564a0085c35d1"
+"checksum want 0.0.5 (registry+https://github.com/rust-lang/crates.io-index)" = "2fffe09593e18ed34950d66dbf44c27deb2e03f3905c493f0641f9f99a3f2349"
 "checksum webpki 0.18.0-alpha4 (registry+https://github.com/rust-lang/crates.io-index)" = "724897af4bb44f3e0142b9cca300eb15f61b9b34fa559440bed8c43f2ff7afc0"
 "checksum which 2.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "49c4f580e93079b70ac522e7bdebbe1568c8afa7d8d05ee534ee737ca37d2f51"
 "checksum widestring 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7157704c2e12e3d2189c507b7482c52820a16dfa4465ba91add92f266667cadb"

diff --git a/proxy/Cargo.toml b/proxy/Cargo.toml
@@ -22,13 +22,14 @@ futures-watch = { git = "https://github.com/carllerche/better-future" }
 h2 = "0.1.10"
 http = "0.1"
 httparse = "1.2"
-hyper = "0.12"
+hyper = "0.12.2"
 ipnet = "1.0"
 log = "0.4.1"
 indexmap = "1.0.0"
 prost = "0.4.0"
 prost-types = "0.4.0"
 rand = "0.5.1"
+try-lock = "0.2"
 
 # for config parsing
 regex = "1.0.0"

diff --git a/proxy/src/lib.rs b/proxy/src/lib.rs
@@ -47,6 +47,7 @@ extern crate conduit_proxy_router;
 extern crate tower_util;
 extern crate tower_in_flight_limit;
 extern crate trust_dns_resolver;
+extern crate try_lock;
 
 use futures::*;
 

diff --git a/proxy/src/task.rs b/proxy/src/task.rs
@@ -3,8 +3,9 @@ use futures::future::{
     Future,
     ExecuteError,
     ExecuteErrorKind,
-    Executor,
 };
+pub use futures::future::Executor;
+
 use tokio::{
     executor::{
         DefaultExecutor,
@@ -20,6 +21,8 @@ use std::{
     io,
 };
 
+pub type BoxSendFuture = Box<Future<Item = (), Error = ()> + Send>;
+
 /// An empty type which implements `Executor` by lazily  calling
 /// `tokio::executor::DefaultExecutor::current().execute(...)`.
 ///
@@ -36,6 +39,12 @@ pub struct LazyExecutor;
 #[derive(Copy, Clone, Debug, Default)]
 pub struct BoxExecutor<E: TokioExecutor>(E);
 
+/// A `futures::executor::Executor` with any generics erased.
+///
+/// This is useful when some code cannot be generic over any executor,
+/// and instead needs a trait object. An example is `Http11Upgrade`.
+pub struct ErasedExecutor(Box<Executor<BoxSendFuture> + Send + Sync>);
+
 /// Indicates which Tokio `Runtime` should be used for the main proxy.
 ///
 /// This is either a `tokio::runtime::current_thread::Runtime`, or a
@@ -71,7 +80,7 @@ pub enum Error {
 impl TokioExecutor for LazyExecutor {
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + 'static + Send>
+        future: BoxSendFuture,
     ) -> Result<(), SpawnError>
     {
         DefaultExecutor::current().spawn(future)
@@ -118,7 +127,7 @@ impl<E: TokioExecutor> BoxExecutor<E> {
 impl<E: TokioExecutor> TokioExecutor for BoxExecutor<E> {
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + 'static + Send>
+        future: BoxSendFuture,
     ) -> Result<(), SpawnError> {
         self.0.spawn(future)
     }
@@ -132,7 +141,7 @@ impl<F, E> Executor<F> for BoxExecutor<E>
 where
     F: Future<Item = (), Error = ()> + 'static + Send,
     E: TokioExecutor,
-    E: Executor<Box<Future<Item = (), Error = ()> + Send + 'static>>,
+    E: Executor<BoxSendFuture>,
 {
     fn execute(&self, future: F) -> Result<(), ExecuteError<F>> {
         // Check the status of the executor first, so that we can return the
@@ -154,6 +163,30 @@ where
     }
 }
 
+// ===== impl ErasedExecutor =====;
+
+impl ErasedExecutor {
+    pub fn erase<E: Executor<BoxSendFuture> + Send + Sync + 'static>(exe: E) -> ErasedExecutor {
+        ErasedExecutor(Box::new(exe))
+    }
+}
+
+impl<F> Executor<F> for ErasedExecutor
+where
+    F: Future<Item = (), Error = ()> + 'static + Send,
+{
+    fn execute(&self, future: F) -> Result<(), ExecuteError<F>> {
+        self.0.execute(Box::new(future))
+            .map_err(|_| panic!("erased executor error"))
+    }
+}
+
+impl fmt::Debug for ErasedExecutor {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.pad("ErasedExecutor")
+    }
+}
+
 // ===== impl MainRuntime =====
 
 impl MainRuntime {

diff --git a/proxy/src/transparency/client.rs b/proxy/src/transparency/client.rs
@@ -11,10 +11,11 @@ use tower_h2;
 use bind;
 use task::BoxExecutor;
 use telemetry::sensor::http::RequestBody;
-use super::glue::{BodyStream, HttpBody, HyperConnect};
+use super::glue::{BodyPayload, HttpBody, HyperConnect};
+use super::upgrade::Http11Upgrade;
 
 type HyperClient<C, B> =
-    hyper::Client<HyperConnect<C>, BodyStream<RequestBody<B>>>;
+    hyper::Client<HyperConnect<C>, BodyPayload<RequestBody<B>>>;
 
 /// A `NewService` that can speak either HTTP/1 or HTTP/2.
 pub struct Client<C, E, B>
@@ -209,12 +210,16 @@ where
     }
 
     fn call(&mut self, req: Self::Request) -> Self::Future {
-        debug!("ClientService::call method={} uri={} headers={:?} ext={:?}",
-            req.method(), req.uri(), req.headers(), req.extensions());
+        debug!("client request: method={} uri={} headers={:?}",
+            req.method(), req.uri(), req.headers());
         match self.inner {
             ClientServiceInner::Http1(ref h1) => {
-                let mut req = hyper::Request::from(req.map(BodyStream::new));
-                ClientServiceFuture::Http1(h1.request(req))
+                let mut req = req.map(BodyPayload::new);
+                let upgrade = req.extensions_mut().remove::<Http11Upgrade>();
+                ClientServiceFuture::Http1 {
+                    future: h1.request(req),
+                    upgrade,
+                }
             },
             ClientServiceInner::Http2(ref mut h2) => {
                 ClientServiceFuture::Http2(h2.call(req))
@@ -224,7 +229,10 @@ where
 }
 
 pub enum ClientServiceFuture {
-    Http1(hyper::client::ResponseFuture),
+    Http1 {
+        future: hyper::client::ResponseFuture,
+        upgrade: Option<Http11Upgrade>,
+    },
     Http2(tower_h2::client::ResponseFuture),
 }
 
@@ -233,12 +241,14 @@ impl Future for ClientServiceFuture {
     type Error = tower_h2::client::Error;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        match *self {
-            ClientServiceFuture::Http1(ref mut f) => {
-                match f.poll() {
+        match self {
+            ClientServiceFuture::Http1 { future, upgrade } => {
+                match future.poll() {
                     Ok(Async::Ready(res)) => {
-                        let res = http::Response::from(res);
-                        let res = res.map(HttpBody::Http1);
+                        let res = res.map(move |b| HttpBody::Http1 {
+                            body: Some(b),
+                            upgrade: upgrade.take(),
+                        });
                         Ok(Async::Ready(res))
                     },
                     Ok(Async::NotReady) => Ok(Async::NotReady),
@@ -248,7 +258,7 @@ impl Future for ClientServiceFuture {
                     }
                 }
             },
-            ClientServiceFuture::Http2(ref mut f) => {
+            ClientServiceFuture::Http2(f) => {
                 let res = try_ready!(f.poll());
                 let res = res.map(HttpBody::Http2);
                 Ok(Async::Ready(res))