Load balance requests to the control plane (#594)

Control plane clients only establish a single connection using the
service's load balancer IP. This presents problems in the face of pod
or node failure.

This change modifies these clients to balance requests across all
available control plane pods. The load balancer is configured by
resolving SRV records to find each pod endpoint. If SRV records are
not present, the client reverts to A-record lookup (as it did previously).

Author: zaharidichev

Cargo.lock

@@ -795,7 +795,6 @@ dependencies = [
 name = "linkerd2-app-core"
 version = "0.1.0"
 dependencies = [
- "async-trait",
  "bytes 0.5.4",
  "futures 0.3.5",
  "html-escape",
@@ -826,6 +825,7 @@ dependencies = [
  "linkerd2-proxy-api-resolve",
  "linkerd2-proxy-core",
  "linkerd2-proxy-discover",
+ "linkerd2-proxy-dns-resolve",
  "linkerd2-proxy-http",
  "linkerd2-proxy-identity",
  "linkerd2-proxy-resolve",
@@ -850,7 +850,6 @@ dependencies = [
  "regex 1.3.9",
  "serde_json",
  "tokio",
- "tokio-test",
  "tokio-timer",
  "tokio-trace",
  "tonic",
@@ -998,6 +997,7 @@ version = "0.1.0"
 dependencies = [
  "futures 0.3.5",
  "linkerd2-dns-name",
+ "linkerd2-error",
  "linkerd2-stack",
  "pin-project",
  "tokio",
@@ -1258,6 +1258,22 @@ dependencies = [
  "tracing-futures",
 ]
 
+[[package]]
+name = "linkerd2-proxy-dns-resolve"
+version = "0.1.0"
+dependencies = [
+ "futures 0.3.5",
+ "linkerd2-addr",
+ "linkerd2-dns",
+ "linkerd2-error",
+ "linkerd2-proxy-core",
+ "tokio",
+ "tokio-test",
+ "tower",
+ "tracing",
+ "tracing-futures",
+]
+
 [[package]]
 name = "linkerd2-proxy-http"
 version = "0.1.0"

Cargo.toml

@@ -31,6 +31,7 @@ members = [
     "linkerd/metrics",
     "linkerd/opencensus",
     "linkerd/proxy/api-resolve",
+    "linkerd/proxy/dns-resolve",
     "linkerd/proxy/core",
     "linkerd/proxy/discover",
     "linkerd/proxy/http",

linkerd/app/core/Cargo.toml

@@ -15,7 +15,6 @@ independently of the inbound and outbound proxy logic.
 mock-orig-dst  = ["linkerd2-proxy-transport/mock-orig-dst"]
 
 [dependencies]
-async-trait = "0.1"
 bytes = "0.5"
 http = "0.2"
 http-body = "0.3"
@@ -47,6 +46,7 @@ linkerd2-proxy-discover = { path = "../../proxy/discover" }
 linkerd2-proxy-identity = { path = "../../proxy/identity" }
 linkerd2-proxy-http = { path = "../../proxy/http" }
 linkerd2-proxy-resolve = { path = "../../proxy/resolve" }
+linkerd2-proxy-dns-resolve = { path = "../../proxy/dns-resolve" }
 linkerd2-proxy-tap = { path = "../../proxy/tap" }
 linkerd2-proxy-tcp = { path = "../../proxy/tcp" }
 linkerd2-proxy-transport = { path = "../../proxy/transport" }
@@ -102,5 +102,3 @@ procinfo = "0.4.2"
 linkerd2-proxy-api = { git = "https://github.com/linkerd/linkerd2-proxy-api", tag = "v0.1.13", features = ["arbitrary"] }
 prost-types = "0.6.0"
 quickcheck = { version = "0.9", default-features = false }
-tokio = { version = "0.2", features = ["time"] }
-tokio-test = "0.2"

linkerd/app/core/src/control.rs

@@ -7,6 +7,12 @@ pub struct ControlAddr {
     pub identity: crate::transport::tls::PeerIdentity,
 }
 
+impl Into<Addr> for ControlAddr {
+    fn into(self) -> Addr {
+        self.addr
+    }
+}
+
 impl fmt::Display for ControlAddr {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         fmt::Display::fmt(&self.addr, f)
@@ -145,174 +151,74 @@ pub mod add_origin {
     }
 }
 
-/// Resolves the controller's `addr` once before building a client.
 pub mod resolve {
-    use super::{client, ControlAddr};
-    use crate::svc;
-    use futures::{ready, TryFuture};
-    use linkerd2_addr::Addr;
-    use linkerd2_dns as dns;
-    use pin_project::pin_project;
-    use std::future::Future;
+    use super::client::Target;
+    use crate::{
+        dns,
+        proxy::{discover, dns_resolve::DnsResolve, resolve::map_endpoint},
+        svc,
+    };
     use std::net::SocketAddr;
-    use std::pin::Pin;
-    use std::task::{Context, Poll};
-    use std::{error, fmt};
 
-    #[derive(Clone, Debug)]
-    pub struct Layer {
+    pub fn layer<M>(
         dns: dns::Resolver,
+    ) -> impl svc::Layer<
+        M,
+        Service = discover::MakeEndpoint<
+            discover::FromResolve<map_endpoint::Resolve<IntoTarget, DnsResolve>, Target>,
+            M,
+        >,
+    > {
+        discover::resolve(map_endpoint::Resolve::new(
+            IntoTarget(()),
+            DnsResolve::new(dns),
+        ))
     }
 
-    #[derive(Clone, Debug)]
-    pub struct Resolve<M> {
-        dns: dns::Resolver,
-        inner: M,
-    }
-
-    #[pin_project]
-    pub struct Init<M>
-    where
-        M: tower::Service<client::Target>,
-    {
-        #[pin]
-        state: State<M>,
-    }
-
-    #[pin_project(project = StateProj)]
-    enum State<M>
-    where
-        M: tower::Service<client::Target>,
-    {
-        Resolve(#[pin] dns::IpAddrFuture, Option<(M, ControlAddr)>),
-        NotReady(M, Option<(SocketAddr, ControlAddr)>),
-        Inner(#[pin] M::Future),
-    }
-
-    #[derive(Debug)]
-    pub enum Error<I> {
-        Dns(dns::Error),
-        Inner(I),
-    }
-
-    // === impl Layer ===
-
-    pub fn layer<M>(dns: dns::Resolver) -> impl svc::Layer<M, Service = Resolve<M>> + Clone
-    where
-        M: tower::Service<client::Target> + Clone,
-    {
-        svc::layer::mk(move |inner| Resolve {
-            dns: dns.clone(),
-            inner,
-        })
-    }
+    #[derive(Copy, Clone, Debug)]
+    pub struct IntoTarget(());
 
-    // === impl Resolve ===
+    impl map_endpoint::MapEndpoint<super::ControlAddr, ()> for IntoTarget {
+        type Out = Target;
 
-    impl<M> tower::Service<ControlAddr> for Resolve<M>
-    where
-        M: tower::Service<client::Target> + Clone,
-    {
-        type Response = M::Response;
-        type Error = <Init<M> as TryFuture>::Error;
-        type Future = Init<M>;
-
-        fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-            self.inner.poll_ready(cx).map_err(Error::Inner)
-        }
-
-        fn call(&mut self, target: ControlAddr) -> Self::Future {
-            let state = match target.addr {
-                Addr::Socket(sa) => State::make_inner(sa, &target, &mut self.inner),
-                Addr::Name(ref na) => {
-                    // The inner service is ready, but we are going to do
-                    // additional work before using it. In case the inner
-                    // service has acquired resources (like a lock), we
-                    // relinquish our claim on the service by replacing it.
-                    self.inner = self.inner.clone();
-
-                    let future = self.dns.resolve_one_ip(na.name());
-                    State::Resolve(future, Some((self.inner.clone(), target.clone())))
-                }
-            };
-
-            Init { state }
-        }
-    }
-
-    // === impl Init ===
-
-    impl<M> Future for Init<M>
-    where
-        M: tower::Service<client::Target>,
-    {
-        type Output = Result<M::Response, Error<M::Error>>;
-
-        fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-            let mut this = self.project();
-            loop {
-                match this.state.as_mut().project() {
-                    StateProj::Resolve(fut, stack) => {
-                        let ip = ready!(fut.poll(cx).map_err(Error::Dns))?;
-                        let (svc, config) = stack.take().unwrap();
-                        let addr = SocketAddr::from((ip, config.addr.port()));
-                        this.state
-                            .as_mut()
-                            .set(State::NotReady(svc, Some((addr, config))));
-                    }
-                    StateProj::NotReady(svc, cfg) => {
-                        ready!(svc.poll_ready(cx).map_err(Error::Inner))?;
-                        let (addr, config) = cfg.take().unwrap();
-                        let state = State::make_inner(addr, &config, svc);
-                        this.state.as_mut().set(state);
-                    }
-                    StateProj::Inner(fut) => return fut.poll(cx).map_err(Error::Inner),
-                };
-            }
+        fn map_endpoint(&self, control: &super::ControlAddr, addr: SocketAddr, _: ()) -> Self::Out {
+            Target::new(addr, control.identity.clone())
         }
     }
+}
 
-    impl<M> State<M>
-    where
-        M: tower::Service<client::Target>,
-    {
-        fn make_inner(addr: SocketAddr, dst: &ControlAddr, mk_svc: &mut M) -> Self {
-            let target = client::Target {
-                addr,
-                server_name: dst.identity.clone(),
-            };
-
-            State::Inner(mk_svc.call(target))
-        }
-    }
+pub mod balance {
+    use crate::proxy::http;
+    use std::time::Duration;
 
-    // === impl Error ===
+    const EWMA_DEFAULT_RTT: Duration = Duration::from_millis(30);
+    const EWMA_DECAY: Duration = Duration::from_secs(10);
 
-    impl<I: fmt::Display> fmt::Display for Error<I> {
-        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-            match self {
-                Error::Dns(dns::Error::NoAddressesFound) => write!(f, "no addresses found"),
-                Error::Dns(e) => fmt::Display::fmt(&e, f),
-                Error::Inner(ref e) => fmt::Display::fmt(&e, f),
-            }
-        }
+    pub fn layer<A, B>() -> http::balance::Layer<A, B> {
+        http::balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY)
     }
-
-    impl<I: fmt::Debug + fmt::Display> error::Error for Error<I> {}
 }
 
 /// Creates a client suitable for gRPC.
 pub mod client {
     use crate::transport::{connect, tls};
     use crate::{proxy::http, svc};
     use linkerd2_proxy_http::h2::Settings as H2Settings;
-    use std::net::SocketAddr;
-    use std::task::{Context, Poll};
+    use std::{
+        net::SocketAddr,
+        task::{Context, Poll},
+    };
 
-    #[derive(Clone, Debug)]
+    #[derive(Clone, Hash, Debug, Eq, PartialEq)]
     pub struct Target {
-        pub(super) addr: SocketAddr,
-        pub(super) server_name: tls::PeerIdentity,
+        addr: SocketAddr,
+        server_name: tls::PeerIdentity,
+    }
+
+    impl Target {
+        pub(super) fn new(addr: SocketAddr, server_name: tls::PeerIdentity) -> Self {
+            Self { addr, server_name }
+        }
     }
 
     #[derive(Debug)]
@@ -356,12 +262,10 @@ pub mod client {
         type Error = <http::h2::Connect<C, B> as tower::Service<Target>>::Error;
         type Future = <http::h2::Connect<C, B> as tower::Service<Target>>::Future;
 
-        #[inline]
         fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
             self.inner.poll_ready(cx)
         }
 
-        #[inline]
         fn call(&mut self, target: Target) -> Self::Future {
             self.inner.call(target)
         }

linkerd/app/core/src/lib.rs

@@ -6,13 +6,14 @@
 //! - Admin interfaces
 //! - Tap
 //! - Metric labeling
-#![type_length_limit = "1586225"]
+
 #![deny(warnings, rust_2018_idioms)]
 
 pub use linkerd2_addr::{self as addr, Addr, NameAddr};
 pub use linkerd2_admit as admit;
 pub use linkerd2_cache as cache;
 pub use linkerd2_conditional::Conditional;
+pub use linkerd2_dns;
 pub use linkerd2_drain as drain;
 pub use linkerd2_error::{Error, Never, Recover};
 pub use linkerd2_exp_backoff as exp_backoff;

linkerd/app/core/src/proxy/mod.rs

@@ -3,6 +3,7 @@
 pub use linkerd2_proxy_api_resolve as api_resolve;
 pub use linkerd2_proxy_core as core;
 pub use linkerd2_proxy_discover as discover;
+pub use linkerd2_proxy_dns_resolve as dns_resolve;
 pub use linkerd2_proxy_http as http;
 pub use linkerd2_proxy_identity as identity;
 pub use linkerd2_proxy_resolve as resolve;

linkerd/app/src/identity.rs

@@ -55,6 +55,7 @@ impl Config {
                     .push_timeout(control.connect.timeout)
                     .push(control::client::layer())
                     .push(control::resolve::layer(dns))
+                    .push_on_response(control::balance::layer())
                     .push(reconnect::layer(Recover(control.connect.backoff)))
                     .push(metrics.into_layer::<classify::Response>())
                     .push(control::add_origin::Layer::new())