Implement reconnect as a NewService (#1032)

In order to make the reconnect module more flexible--so that it can more
easily be used in different parts of the stack--this change modifies the
`reconnect` crate to be implemented in terms of `NewService` rather than
a `MakeService`.

Furthermore, tests are added to exercise reconnect.

This change sets up changes to reconnect behavior so reconnection is
only used when service discovery is responsible for evicting endpoints
from a balancer.

Author: olix0r

Cargo.lock

@@ -1289,8 +1289,13 @@ dependencies = [
  "futures",
  "linkerd-error",
  "linkerd-stack",
+ "linkerd-tracing",
  "pin-project",
+ "tokio",
+ "tokio-stream",
+ "tokio-test",
  "tower",
+ "tower-test",
  "tracing",
 ]
 

linkerd/app/core/src/control.rs

@@ -1,6 +1,6 @@
 use crate::{
-    classify, config, control, dns, metrics, proxy::http, reconnect, svc, tls,
-    transport::ConnectTcp, Addr, Error,
+    classify, config, control, dns, metrics, proxy::http, svc, tls, transport::ConnectTcp, Addr,
+    Error,
 };
 use futures::future::Either;
 use std::fmt;
@@ -55,11 +55,6 @@ impl Config {
     {
         let addr = self.addr;
 
-        let connect_backoff = {
-            let backoff = self.connect.backoff;
-            move |_| Ok(backoff.stream())
-        };
-
         // When a DNS resolution fails, log the error and use the TTL, if there
         // is one, to drive re-resolution attempts.
         let resolve_backoff = {
@@ -89,7 +84,9 @@ impl Config {
             .push(tls::Client::layer(identity))
             .push_timeout(self.connect.timeout)
             .push(self::client::layer())
-            .push(reconnect::layer(connect_backoff))
+            .push_on_response(svc::MapErrLayer::new(Into::into))
+            .into_new_service()
+            .push_new_reconnect(self.connect.backoff)
             // Ensure individual endpoints are driven to readiness so that the balancer need not
             // drive them all directly.
             .push_on_response(svc::layer::mk(svc::SpawnReady::new))

linkerd/app/core/src/lib.rs

@@ -23,7 +23,6 @@ pub use linkerd_http_metrics as http_metrics;
 pub use linkerd_identity as identity;
 pub use linkerd_io as io;
 pub use linkerd_opencensus as opencensus;
-pub use linkerd_reconnect as reconnect;
 pub use linkerd_service_profiles as profiles;
 pub use linkerd_stack_metrics as stack_metrics;
 pub use linkerd_stack_tracing as stack_tracing;

linkerd/app/core/src/svc.rs

@@ -3,6 +3,9 @@
 pub use crate::proxy::http;
 use crate::{cache, Error};
 pub use linkerd_concurrency_limit::ConcurrencyLimit;
+use linkerd_error::Recover;
+use linkerd_exp_backoff::{ExponentialBackoff, ExponentialBackoffStream};
+pub use linkerd_reconnect::NewReconnect;
 pub use linkerd_stack::{
     self as stack, layer, BoxNewService, BoxService, BoxServiceLayer, Fail, Filter, MapTargetLayer,
     NewRouter, NewService, Param, Predicate, UnwrapOr,
@@ -26,6 +29,9 @@ pub use tower::{
     Service, ServiceExt,
 };
 
+#[derive(Copy, Clone, Debug)]
+pub struct AlwaysReconnect(ExponentialBackoff);
+
 pub type Buffer<Req, Rsp, E> = TowerBuffer<BoxService<Req, Rsp, E>, Req>;
 
 pub type BoxHttp<B = http::BoxBody> =
@@ -47,6 +53,8 @@ pub fn stack<S>(inner: S) -> Stack<S> {
     Stack(inner)
 }
 
+// === impl IdentityProxy ===
+
 pub fn proxies() -> Stack<IdentityProxy> {
     Stack(IdentityProxy(()))
 }
@@ -59,6 +67,8 @@ impl<T> NewService<T> for IdentityProxy {
     fn new_service(&mut self, _: T) -> Self::Service {}
 }
 
+// === impl Layers ===
+
 #[allow(dead_code)]
 impl<L> Layers<L> {
     pub fn push<O>(self, outer: O) -> Layers<Pair<L, O>> {
@@ -69,13 +79,6 @@ impl<L> Layers<L> {
         self.push(stack::MapTargetLayer::new(map_target))
     }
 
-    /// Wraps an inner `MakeService` to be a `NewService`.
-    pub fn push_into_new_service(
-        self,
-    ) -> Layers<Pair<L, stack::new_service::FromMakeServiceLayer>> {
-        self.push(stack::new_service::FromMakeServiceLayer::default())
-    }
-
     /// Buffers requests in an mpsc, spawning the inner service onto a dedicated task.
     pub fn push_spawn_buffer<Req>(
         self,
@@ -105,6 +108,8 @@ impl<M, L: Layer<M>> Layer<M> for Layers<L> {
     }
 }
 
+// === impl Stack ===
+
 #[allow(dead_code)]
 impl<S> Stack<S> {
     pub fn push<L: Layer<S>>(self, layer: L) -> Stack<L::Service> {
@@ -137,7 +142,14 @@ impl<S> Stack<S> {
 
     /// Wraps an inner `MakeService` to be a `NewService`.
     pub fn into_new_service(self) -> Stack<stack::new_service::FromMakeService<S>> {
-        self.push(stack::new_service::FromMakeServiceLayer::default())
+        self.push(stack::new_service::FromMakeService::layer())
+    }
+
+    pub fn push_new_reconnect(
+        self,
+        backoff: ExponentialBackoff,
+    ) -> Stack<NewReconnect<AlwaysReconnect, S>> {
+        self.push(NewReconnect::layer(AlwaysReconnect(backoff)))
     }
 
     /// Buffer requests when when the next layer is out of capacity.
@@ -331,3 +343,13 @@ where
         self.0.call(t)
     }
 }
+
+// === impl AlwaysReconnect ===
+
+impl<E: Into<Error>> Recover<E> for AlwaysReconnect {
+    type Backoff = ExponentialBackoffStream;
+
+    fn recover(&self, _: E) -> Result<Self::Backoff, E> {
+        Ok(self.0.stream())
+    }
+}

linkerd/app/inbound/src/http/mod.rs

@@ -17,7 +17,6 @@ use linkerd_app_core::{
     config::{ProxyConfig, ServerConfig},
     dst, errors, http_tracing, identity, io, profiles,
     proxy::{http, tap},
-    reconnect,
     svc::{self, Param},
     Error,
 };
@@ -140,10 +139,9 @@ where
                 config.proxy.connect.h1_settings,
                 config.proxy.connect.h2_settings,
             ))
-            .push(reconnect::layer({
-                let backoff = config.proxy.connect.backoff;
-                move |_| Ok(backoff.stream())
-            }))
+            .push_on_response(svc::MapErrLayer::new(Into::into))
+            .into_new_service()
+            .push_new_reconnect(config.proxy.connect.backoff)
             .check_new_service::<HttpEndpoint, http::Request<_>>();
 
         let target = endpoint

linkerd/app/outbound/src/http/endpoint.rs

@@ -3,7 +3,7 @@ use crate::Outbound;
 use linkerd_app_core::{
     classify, config, http_tracing, metrics,
     proxy::{http, tap},
-    reconnect, svc, tls, Error, CANONICAL_DST_HEADER,
+    svc, tls, Error, CANONICAL_DST_HEADER,
 };
 use tokio::io;
 
@@ -40,12 +40,10 @@ impl<C> Outbound<C> {
         // HTTP/1.x fallback is supported as needed.
         let stack = connect
             .push(http::client::layer(h1_settings, h2_settings))
+            .push_on_response(svc::MapErrLayer::new(Into::<Error>::into))
             .check_service::<T>()
-            // Re-establishes a connection when the client fails.
-            .push(reconnect::layer({
-                let backoff = backoff;
-                move |_| Ok(backoff.stream())
-            }))
+            .into_new_service()
+            .push_new_reconnect(backoff)
             .push(tap::NewTapHttp::layer(rt.tap.clone()))
             .push(rt.metrics.http_endpoint.to_layer::<classify::Response, _>())
             .push_on_response(http_tracing::client(

linkerd/reconnect/Cargo.toml

@@ -13,3 +13,10 @@ futures = { version = "0.3", default-features = false }
 tower = { version = "0.4.7", default-features = false }
 tracing = "0.1.23"
 pin-project = "1"
+
+[dev-dependencies]
+linkerd-tracing = { path = "../tracing" }
+tokio = { version = "1", features = ["macros", "rt", "time"] }
+tokio-stream = { version = "0.1", features = ["time"] }
+tokio-test = "0.4"
+tower-test = "0.4"