outbound: Refactor stack target types (#2210)

To support a new configuration API, we want to update the outbound stack
to compose the following stacks:

* Routing discovery;
* Protocol switching;
* Protocol-specific logical routing;
* Concrete forwarding / load balancing;
* Endpoint dispatching.

To do this, we update the existing stacks so that:

* each stack accepts generic input targets (with enumerated `Param`
  requirements); and
* each module defines all necessary internal target types, including
  those passed to inner stacks.

This decouples stack configuration from the specifics of discovery:
tests can satisfy stack parameters without having to stub out all of the
discovery types and we are setup to change discovery implementation
details moving forward.

This requires updating the gateway and ingress stacks in addition to the
(default) sidecar stack.

Generally, we prefer duplicating similar target types to coupling
modules. In followup changes, we can introduce macros to reduce `Param`
boilerplate if necessary.

Changes include:

* A new `Gateway` builder type (analogous to `Inbound` and `Outbound`);
* A new `Gateway::server` stack handles discovery and protocol
  switching;
* Service cacheing is performed for the http and opaq stacks (to
  preserve load balancer state, etc) using only the information needed
  for those stacks;
* Concrete stacks provide the only data-plane queue. This enables
  outer stacks to be cloneable. This also means that the concrete stacks
  are the only stacks with stack metrics;
* The `switch_logical` module is removed;
* The `http::detect` module is moved into the `protocol` stack, which
  now supports discovery-hinted protocols to bypass detection.
* The `logical` and `endpoint` modules are removed in favor of
  protocol-specific modules.

Author: olix0r

linkerd/app/gateway/src/http.rs

@@ -1,54 +1,169 @@
-use self::gateway::NewHttpGateway;
-use super::OutboundHttp;
-use linkerd_app_core::{
-    identity, io,
-    proxy::{api_resolve::Metadata, core::Resolve, http},
-    svc,
-    transport::{ClientAddr, Local},
-    Error, Infallible,
+use super::{server::Http, Gateway};
+use inbound::{GatewayAddr, GatewayDomainInvalid};
+use linkerd_app_core::{identity, io, profiles, proxy::http, svc, tls, transport::addrs::*, Error};
+use linkerd_app_inbound as inbound;
+use linkerd_app_outbound as outbound;
+use std::{
+    cmp::{Eq, PartialEq},
+    fmt::Debug,
+    hash::Hash,
 };
-use linkerd_app_outbound::{self as outbound, Outbound};
 
 mod gateway;
 #[cfg(test)]
 mod tests;
 
-pub use linkerd_app_core::proxy::http::*;
+pub(crate) use self::gateway::NewHttpGateway;
 
-/// Builds an outbound HTTP stack.
+/// Target for outbound HTTP gateway stacks.
+#[derive(Clone, Debug, PartialEq, Eq, Hash)]
+pub struct Target<T = ()> {
+    addr: GatewayAddr,
+    target: outbound::http::Logical,
+    version: http::Version,
+    parent: T,
+}
+
+/// Implements `svc::router::SelectRoute` for outbound HTTP requests. An
+/// `OutboundHttp` target is returned for each request using the request's HTTP
+/// version.
 ///
-/// A gateway-specififc module is inserted to requests from looping through
-/// gateways. Discovery errors are lifted into the HTTP stack so that individual
-/// requests are failed with an HTTP-level error repsonse.
-pub(super) fn stack<O, R>(
-    local_id: identity::LocalId,
-    outbound: Outbound<O>,
-    resolve: R,
-) -> svc::ArcNewService<
-    OutboundHttp,
-    impl svc::Service<
-        http::Request<http::BoxBody>,
-        Response = http::Response<http::BoxBody>,
-        Error = Error,
-        Future = impl Send,
-    >,
->
+/// The request's HTTP version may not match the target's original HTTP version
+/// when proxies use HTTP/2 to transport HTTP/1 requests.
+#[derive(Clone, Debug)]
+struct ByRequestVersion<T>(Target<T>);
+
+impl Gateway {
+    /// Wrap the provided outbound HTTP stack with an HTTP server, inbound
+    /// authorization, and gateway request routing.
+    pub fn http<T, I, N, NSvc>(&self, inner: N) -> svc::Stack<svc::ArcNewTcp<Http<T>, I>>
+    where
+        // Target describing an inbound gateway connection.
+        T: svc::Param<GatewayAddr>,
+        T: svc::Param<OrigDstAddr>,
+        T: svc::Param<Remote<ClientAddr>>,
+        T: svc::Param<tls::ConditionalServerTls>,
+        T: svc::Param<tls::ClientId>,
+        T: svc::Param<inbound::policy::AllowPolicy>,
+        T: svc::Param<profiles::LookupAddr>,
+        T: Clone + Send + Sync + Unpin + 'static,
+        // Server-side socket.
+        I: io::AsyncRead + io::AsyncWrite + io::PeerAddr,
+        I: Send + Unpin + 'static,
+        // HTTP outbound stack.
+        N: svc::NewService<Target, Service = NSvc>,
+        N: Clone + Send + Sync + Unpin + 'static,
+        NSvc: svc::Service<
+            http::Request<http::BoxBody>,
+            Response = http::Response<http::BoxBody>,
+            Error = Error,
+        >,
+        NSvc: Send + Unpin + 'static,
+        NSvc::Future: Send + 'static,
+    {
+        let http = svc::stack(inner)
+            // Discard `T` and its associated client-specific metadata.
+            .push_map_target(Target::discard_parent)
+            // Add headers to prevent loops.
+            .push(NewHttpGateway::layer(identity::LocalId(
+                self.inbound.identity().name().clone(),
+            )))
+            .push_on_service(svc::LoadShed::layer())
+            .lift_new()
+            // After protocol-downgrade, we need to build an inner stack for
+            // each request-level HTTP version.
+            .push(svc::NewOneshotRoute::layer_via(|t: &Target<T>| {
+                ByRequestVersion(t.clone())
+            }))
+            // Only permit gateway traffic to endpoints for which we have
+            // discovery information.
+            .push_filter(
+                |(_, parent): (_, Http<T>)| -> Result<_, GatewayDomainInvalid> {
+                    let target = {
+                        let profile = svc::Param::<Option<profiles::Receiver>>::param(&parent)
+                            .ok_or(GatewayDomainInvalid)?;
+
+                        if let Some(profiles::LogicalAddr(addr)) = profile.logical_addr() {
+                            outbound::http::Logical::Route(addr, profile)
+                        } else if let Some((addr, metadata)) = profile.endpoint() {
+                            outbound::http::Logical::Forward(Remote(ServerAddr(addr)), metadata)
+                        } else {
+                            return Err(GatewayDomainInvalid);
+                        }
+                    };
+
+                    Ok(Target {
+                        target,
+                        addr: (*parent).param(),
+                        version: svc::Param::param(&parent),
+                        parent: (**parent).clone(),
+                    })
+                },
+            )
+            // Authorize requests to the gateway.
+            .push(self.inbound.authorize_http());
+
+        self.inbound
+            .clone()
+            .with_stack(http.into_inner())
+            // Teminates HTTP connections.
+            // XXX Sets an identity header -- this should probably not be done
+            // in the gateway, though the value will be stripped by meshed
+            // servers.
+            .push_http_server()
+            .into_stack()
+    }
+}
+
+// === impl ByRequestVersion ===
+
+impl<B, T: Clone> svc::router::SelectRoute<http::Request<B>> for ByRequestVersion<T> {
+    type Key = Target<T>;
+    type Error = http::version::Unsupported;
+
+    fn select(&self, req: &http::Request<B>) -> Result<Self::Key, Self::Error> {
+        let mut t = self.0.clone();
+        t.version = req.version().try_into()?;
+        Ok(t)
+    }
+}
+
+// === impl Target ===
+
+impl<T> Target<T> {
+    fn discard_parent(self) -> Target {
+        Target {
+            addr: self.addr,
+            target: self.target,
+            version: self.version,
+            parent: (),
+        }
+    }
+}
+
+impl<T> svc::Param<GatewayAddr> for Target<T> {
+    fn param(&self) -> GatewayAddr {
+        self.addr.clone()
+    }
+}
+
+impl<T> svc::Param<http::Version> for Target<T> {
+    fn param(&self) -> http::Version {
+        self.version
+    }
+}
+
+impl<T> svc::Param<tls::ClientId> for Target<T>
 where
-    O: Clone + Send + Sync + Unpin + 'static,
-    O: svc::MakeConnection<outbound::tcp::Connect, Metadata = Local<ClientAddr>, Error = io::Error>,
-    O::Connection: Send + Unpin,
-    O::Future: Send + Unpin + 'static,
-    R: Resolve<outbound::http::Concrete, Endpoint = Metadata, Error = Error>,
+    T: svc::Param<tls::ClientId>,
 {
-    let endpoint = outbound.push_tcp_endpoint().push_http_endpoint();
-    endpoint
-        .clone()
-        .push_http_concrete(resolve)
-        .push_http_logical()
-        .into_stack()
-        .push_switch(Ok::<_, Infallible>, endpoint.into_stack())
-        .push(NewHttpGateway::layer(local_id))
-        .push(svc::ArcNewService::layer())
-        .check_new::<OutboundHttp>()
-        .into_inner()
+    fn param(&self) -> tls::ClientId {
+        self.parent.param()
+    }
+}
+
+impl<T> svc::Param<outbound::http::Logical> for Target<T> {
+    fn param(&self) -> outbound::http::Logical {
+        self.target.clone()
+    }
 }