Update routers to support per-request backend distributions

Cargo.lock

@@ -939,14 +939,14 @@ dependencies = [
  "hyper",
  "linkerd-app-core",
  "linkerd-app-test",
+ "linkerd-distribute",
  "linkerd-http-classify",
  "linkerd-http-retry",
  "linkerd-identity",
  "linkerd-io",
  "linkerd-meshtls",
  "linkerd-meshtls-rustls",
  "linkerd-retry",
- "linkerd-router",
  "linkerd-tracing",
  "parking_lot",
  "pin-project",
@@ -1547,7 +1547,6 @@ dependencies = [
  "futures",
  "http",
  "http-body",
- "indexmap",
  "linkerd-addr",
  "linkerd-dns-name",
  "linkerd-error",
@@ -1557,10 +1556,8 @@ dependencies = [
  "linkerd-tonic-watch",
  "linkerd2-proxy-api",
  "once_cell",
- "pin-project",
  "prost-types",
  "quickcheck",
- "rand",
  "regex",
  "thiserror",
  "tokio",

linkerd/app/outbound/Cargo.toml

@@ -19,11 +19,11 @@ bytes = "1"
 http = "0.2"
 futures = { version = "0.3", default-features = false }
 linkerd-app-core = { path = "../core" }
+linkerd-distribute = { path = "../../distribute" }
 linkerd-http-classify = { path = "../../http-classify" }
 linkerd-http-retry = { path = "../../http-retry" }
 linkerd-identity = { path = "../../identity" }
 linkerd-retry = { path = "../../retry" }
-linkerd-router = { path = "../../router" }
 parking_lot = "0.12"
 thiserror = "1"
 tokio = { version = "1", features = ["sync"] }

linkerd/app/outbound/src/endpoint.rs

@@ -1,4 +1,4 @@
-use crate::{http, logical::Concrete, tcp, Outbound};
+use crate::{http, logical::Concrete, stack_labels, tcp, Outbound};
 use linkerd_app_core::{
     io, metrics,
     profiles::LogicalAddr,
@@ -13,6 +13,7 @@ use std::{
     net::{IpAddr, SocketAddr},
     sync::Arc,
 };
+use tracing::info_span;
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Endpoint<P> {
@@ -201,7 +202,14 @@ impl<P: Copy + std::fmt::Debug> MapEndpoint<Concrete<P>, Metadata> for FromMetad
 // === Outbound ===
 
 impl<S> Outbound<S> {
-    pub fn push_endpoint<I>(self) -> Outbound<svc::ArcNewTcp<tcp::Endpoint, I>>
+    /// Builds a stack that handles forwarding a connection to a single endpoint
+    /// (i.e. without routing and load balancing).
+    ///
+    /// HTTP protocol detection may still be performed on the connection.
+    ///
+    /// A service produced by this stack is used for a single connection (i.e.
+    /// without any form of caching for reuse across connections).
+    pub fn push_forward<I>(self) -> Outbound<svc::ArcNewTcp<tcp::Endpoint, I>>
     where
         Self: Clone + 'static,
         S: svc::MakeConnection<tcp::Connect, Metadata = Local<ClientAddr>, Error = io::Error>,
@@ -211,19 +219,38 @@ impl<S> Outbound<S> {
         I: io::AsyncRead + io::AsyncWrite + io::PeerAddr,
         I: fmt::Debug + Send + Sync + Unpin + 'static,
     {
+        // The forwarding stacks are **NOT** cached, since they don't do any
+        // external discovery.
         let http = self
             .clone()
             .push_tcp_endpoint::<http::Connect>()
             .push_http_endpoint()
-            .map_stack(|config, _, stk| {
-                stk.push_buffer_on_service("HTTP Server", &config.http_request_buffer)
+            .map_stack(|config, rt, stk| {
+                stk.push_on_service(
+                    svc::layers()
+                        .push(
+                            rt.metrics
+                                .proxy
+                                .stack
+                                .layer(stack_labels("http", "forward")),
+                        )
+                        // TODO(ver): This buffer config should be distinct from
+                        // that in the concrete stack. It should probably be
+                        // derived from the target so that we can configure it
+                        // via the API.
+                        .push_buffer("HTTP Forward", &config.http_request_buffer),
+                )
             })
             .push_http_server()
             .into_inner();
 
-        self.push_tcp_endpoint()
-            .push_tcp_forward()
-            .push_detect_http(http)
+        let opaque = self.push_tcp_endpoint().push_tcp_forward();
+
+        opaque.push_detect_http(http).map_stack(|_, _, stk| {
+            stk.instrument(|e: &tcp::Endpoint| info_span!("forward", endpoint = %e.addr))
+                .push_on_service(svc::BoxService::layer())
+                .push(svc::ArcNewService::layer())
+        })
     }
 }
 
@@ -256,7 +283,7 @@ pub mod tests {
                         "i don't like you, go away",
                     ))
                 }))
-                .push_endpoint()
+                .push_forward()
                 .into_inner()
                 .new_service(tcp::Endpoint::forward(
                     OrigDstAddr(addr),

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

@@ -9,20 +9,30 @@ use linkerd_app_core::{
     },
     svc, Error, Infallible,
 };
+use tracing::info_span;
 
 impl<N> Outbound<N> {
+    /// Builds a [`svc::NewService`] stack that builds buffered HTTP load
+    /// balancer services for [`Concrete`] targets.
+    ///
+    /// When a balancer has no available inner services, it goes into
+    /// 'failfast'. While in failfast, buffered requests are failed and the
+    /// service becomes unavailable so callers may choose alternate concrete
+    /// services.
+    //
+    // TODO(ver) make the outer target type generic/parameterized.
     pub fn push_http_concrete<NSvc, R>(
         self,
         resolve: R,
     ) -> Outbound<
         svc::ArcNewService<
             Concrete,
             impl svc::Service<
-                http::Request<http::BoxBody>,
-                Response = http::Response<http::BoxBody>,
-                Error = Error,
-                Future = impl Send,
-            >,
+                    http::Request<http::BoxBody>,
+                    Response = http::Response<http::BoxBody>,
+                    Error = Error,
+                    Future = impl Send,
+                > + Clone,
         >,
     >
     where
@@ -38,14 +48,7 @@ impl<N> Outbound<N> {
         R::Future: Send + Unpin,
     {
         self.map_stack(|config, rt, endpoint| {
-            let crate::Config {
-                discovery_idle_timeout,
-                ..
-            } = config;
-            let watchdog = *discovery_idle_timeout * 2;
-
             let resolve = svc::stack(resolve.into_service())
-                .check_service::<ConcreteAddr>()
                 .push_request_filter(|c: Concrete| Ok::<_, Infallible>(c.resolve))
                 .push(svc::layer::mk(move |inner| {
                     map_endpoint::Resolve::new(
@@ -59,51 +62,43 @@ impl<N> Outbound<N> {
                 .into_inner();
 
             endpoint
-                .check_new_service::<Endpoint, http::Request<http::BoxBody>>()
                 .push_on_service(
-                    svc::layers().push(http::BoxRequest::layer()).push(
-                        rt.metrics
-                            .proxy
-                            .stack
-                            .layer(stack_labels("http", "balance.endpoint")),
-                    ),
+                    rt.metrics
+                        .proxy
+                        .stack
+                        .layer(stack_labels("http", "endpoint")),
                 )
-                .instrument(|t: &Endpoint| tracing::debug_span!("endpoint", addr = %t.addr))
-                .check_new_service::<Endpoint, http::Request<_>>()
+                .instrument(|e: &Endpoint| info_span!("endpoint", addr = %e.addr))
                 // Resolve the service to its endpoints and balance requests over them.
                 //
-                // If the balancer has been empty/unavailable, eagerly fail requests.
-                // When the balancer is in failfast, spawn the service in a background
-                // task so it becomes ready without new requests.
-                //
                 // We *don't* ensure that the endpoint is driven to readiness here, because this
                 // might cause us to continually attempt to reestablish connections without
                 // consulting discovery to see whether the endpoint has been removed. Instead, the
-                // endpoint layer spawns each _connection_ attempt on a background task, but the
+                // endpoint stack spawns each _connection_ attempt on a background task, but the
                 // decision to attempt the connection must be driven by the balancer.
-                .push(resolve::layer(resolve, watchdog))
+                //
+                // TODO(ver) remove the watchdog timeout.
+                .push(resolve::layer(resolve, config.discovery_idle_timeout * 2))
+                .push_on_service(http::balance::layer(
+                    crate::EWMA_DEFAULT_RTT,
+                    crate::EWMA_DECAY,
+                ))
+                .check_make_service::<Concrete, http::Request<http::BoxBody>>()
+                .push(svc::MapErr::layer(Into::into))
+                // Drives the initial resolution via the service's readiness.
+                .into_new_service()
                 .push_on_service(
                     svc::layers()
-                        .push(http::balance::layer(
-                            crate::EWMA_DEFAULT_RTT,
-                            crate::EWMA_DECAY,
-                        ))
+                        .push(http::BoxResponse::layer())
                         .push(
                             rt.metrics
                                 .proxy
                                 .stack
-                                .layer(stack_labels("http", "balancer")),
+                                .layer(stack_labels("http", "concrete")),
                         )
-                        .push(http::BoxResponse::layer()),
+                        .push_buffer("HTTP Concrete", &config.http_request_buffer),
                 )
-                .check_make_service::<Concrete, http::Request<_>>()
-                .push(svc::MapErr::layer(Into::into))
-                // Drives the initial resolution via the service's readiness.
-                .into_new_service()
-                // The concrete address is only set when the profile could be
-                // resolved. Endpoint resolution is skipped when there is no
-                // concrete address.
-                .instrument(|c: &Concrete| tracing::debug_span!("concrete", service = %c.resolve))
+                .instrument(|c: &Concrete| info_span!("concrete", svc = %c.resolve))
                 .push(svc::ArcNewService::layer())
         })
     }

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

@@ -11,6 +11,18 @@ use tracing::debug_span;
 pub struct Skip;
 
 impl<N> Outbound<N> {
+    /// Builds a `NewService` that produces services that optionally (depending
+    /// on the target) perform HTTP protocol detection on sockets.
+    ///
+    /// When HTTP is detected, an HTTP service is build from the provided HTTP
+    /// stack. In either case, the inner service is built for each connection so
+    /// inner services must implement caching as needed.
+    //
+    // TODO(ver) We can be smarter about reusing inner services across
+    // connections by moving caching into this stack...
+    //
+    // TODO(ver) Let discovery influence whether we assume an HTTP protocol
+    // without deteciton.
     pub fn push_detect_http<T, U, NSvc, H, HSvc, I>(self, http: H) -> Outbound<svc::ArcNewTcp<T, I>>
     where
         I: io::AsyncRead + io::AsyncWrite + io::PeerAddr,
@@ -31,6 +43,8 @@ impl<N> Outbound<N> {
         self.map_stack(|config, rt, tcp| {
             let ServerConfig { h2_settings, .. } = config.proxy.server;
 
+            let tcp = tcp.instrument(|_: &_| debug_span!("opaque"));
+
             svc::stack(http)
                 .push_on_service(
                     svc::layers()
@@ -46,7 +60,14 @@ impl<N> Outbound<N> {
                         .push_on_service(svc::MapTargetLayer::new(io::EitherIo::Right))
                         .into_inner(),
                 ))
-                .push_on_service(svc::BoxService::layer())
+                .push_on_service(
+                    svc::layers()
+                        // `DetectService` oneshots the inner service, so we add
+                        // a loadshed to prevent leaking tasks if (for some
+                        // unexpected reason) the inner service is not ready.
+                        .push(svc::LoadShed::layer())
+                        .push(svc::BoxService::layer()),
+                )
                 .check_new_service::<(Option<http::Version>, T), _>()
                 .push_map_target(detect::allow_timeout)
                 .push(svc::ArcNewService::layer())