Update routers to support per-request backend distributions (#2095)

The outbound logical stacks use a single `profiles::split` for all
requests. Therefore, request routing can't influence a request's backend
selection.

This change replaces the service-profile-specific router and split
implementations in favor of the newer, more general router (577ad52) and
distribute (47678e6) modules. Unlike the service-profile-router, the
generic router is lazy and only creates route services as a matching
request is found.

Buffers are moved into the concrete stack so that balancers may be
shared by more than one route.

This change also updates trace contexts so that we include
logical/endpoint metadata as early as possible.

Co-authored-by: Eliza Weisman <[email protected]>

Author: olix0r

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())