proxy: Rebind endpoint stacks on TLS config changes using WatchService (#1185)

This branch changes the proxy's `Bind` module to add a middleware layer 
which watches for TLS cliend configuration changes and rebinds the 
endpoint stacks of any endpoints with which it is able to communicate with over 
TLS (i.e. those with `TlsIdentity` metadata) when the client config changes. The 
rebinding is done at the level of individual endpoint stacks, rather than for the
entire service stack for the destination.

This obsoletes my previous PRs #1169 and #1175.

Closes #1161

Signed-off-by: Eliza Weisman <eliza@buoyant.io>

proxy/src/bind.rs

@@ -19,6 +19,7 @@ use transport;
 use tls;
 use ctx::transport::TlsStatus;
 use conditional::Conditional;
+use watch_service::{WatchService, Rebind};
 
 /// Binds a `Service` from a `SocketAddr`.
 ///
@@ -74,7 +75,7 @@ where
     B: tower_h2::Body + Send + 'static,
     <B::Data as ::bytes::IntoBuf>::Buf: Send,
 {
-    Bound(Stack<B>),
+    Bound(WatchService<Option<tls::ClientConfig>, RebindTls<B>>),
     BindsPerRequest {
         // When `poll_ready` is called, the _next_ service to be used may be bound
         // ahead-of-time. This stack is used only to serve the next request to this
@@ -123,9 +124,17 @@ pub struct NormalizeUri<S> {
     was_absolute_form: bool,
 }
 
+pub struct RebindTls<B> {
+    bind: Bind<Arc<ctx::Proxy>, B>,
+    protocol: Protocol,
+    endpoint: Endpoint,
+}
+
 pub type Service<B> = BoundService<B>;
 
-pub type Stack<B> = Reconnect<NormalizeUri<NewHttp<B>>>;
+pub type Stack<B> = WatchService<Option<tls::ClientConfig>, RebindTls<B>>;
+
+type StackInner<B> = Reconnect<NormalizeUri<NewHttp<B>>>;
 
 pub type NewHttp<B> = sensor::NewHttp<Client<B>, B, HttpBody>;
 
@@ -205,13 +214,27 @@ where
     B: tower_h2::Body + Send + 'static,
     <B::Data as ::bytes::IntoBuf>::Buf: Send,
 {
-    fn bind_stack(&self, ep: &Endpoint, protocol: &Protocol) -> Stack<B> {
-        debug!("bind_stack endpoint={:?}, protocol={:?}", ep, protocol);
+    /// Binds the "inner" layers of the stack.
+    ///
+    /// This binds a service stack that comprises the client for that individual
+    /// endpoint. It will have to be rebuilt if the TLS configuration changes.
+    ///
+    /// This includes:
+    /// + Reconnects
+    /// + URI normalization
+    /// + HTTP sensors
+    ///
+    /// When the TLS client configuration is invalidated, this function will
+    /// be called again to bind a new stack.
+    fn bind_inner_stack(&self, ep: &Endpoint, protocol: &Protocol)-> StackInner<B> {
+        debug!("bind_inner_stack endpoint={:?}, protocol={:?}", ep, protocol);
         let addr = ep.address();
 
         // Like `tls::current_connection_config()` with optional identity.
         let tls = match ep.tls_identity() {
             Conditional::Some(identity) => {
+                // TODO: the watch should be an explicit field of `Bind`, rather
+                // than passed in the context.
                 match *self.ctx.tls_client_config_watch().borrow() {
                     Some(ref config) =>
                         Conditional::Some(tls::ConnectionConfig {
@@ -260,6 +283,34 @@ where
         Reconnect::new(proxy)
     }
 
+    /// Binds the endpoint stack used to construct a bound service.
+    ///
+    /// This will wrap the service stack returned by `bind_inner_stack`
+    /// with a middleware layer that causes it to be re-constructed when
+    /// the TLS client configuration changes.
+    ///
+    /// This function will itself be called again by `BoundService` if the
+    /// service binds per request, or if the initial connection to the
+    /// endpoint fails.
+    fn bind_stack(&self, ep: &Endpoint, protocol: &Protocol) -> Stack<B> {
+        debug!("bind_stack: endpoint={:?}, protocol={:?}", ep, protocol);
+        // TODO: Since `BindsPerRequest` bindings are only used for a
+        // single request, it seems somewhat unnecessary to wrap them in a
+        // `WatchService` middleware so that they can be rebound when the TLS
+        // config changes, since they _always_ get rebound regardless. For now,
+        // we still add the `WatchService` layer so that the per-request and
+        // bound service stacks have the same type.
+        let rebind = RebindTls {
+            bind: self.clone(),
+            endpoint: ep.clone(),
+            protocol: protocol.clone(),
+        };
+        // TODO: the watch should be an explicit field of `Bind`, rather
+        // than passed in the context.
+        let tls_client_cfg = self.ctx.tls_client_config_watch().clone();
+        WatchService::new(tls_client_cfg, rebind)
+    }
+
     pub fn bind_service(&self, ep: &Endpoint, protocol: &Protocol) -> BoundService<B> {
         let binding = if protocol.can_reuse_clients() {
             Binding::Bound(self.bind_stack(ep, protocol))
@@ -524,3 +575,22 @@ impl Protocol {
         }
     }
 }
+
+// ===== impl RebindTls =====
+
+impl<B> Rebind<Option<tls::ClientConfig>> for RebindTls<B>
+where
+    B: tower_h2::Body + Send + 'static,
+    <B::Data as ::bytes::IntoBuf>::Buf: Send,
+{
+    type Service = StackInner<B>;
+    fn rebind(&mut self, _cfg: &Option<tls::ClientConfig>) -> Self::Service {
+        debug!(
+            "rebinding endpoint stack for {:?}:{:?} on TLS config change",
+            self.endpoint, self.protocol,
+        );
+        // We don't actually pass in the config, as `self.bind` also already
+        // owns a config watch of its own.
+        self.bind.bind_inner_stack(&self.endpoint, &self.protocol)
+    }
+}