gateway: Fix URI form for gatewayed HTTP/1 requests (#2241)

A recent change broke how gateways handle the URI of upgraded HTTP/1.1
requests. This problem was caused by normaling the URI twice (once in
the inbound HTTP server and once in the outbound HTTP server).

To fix this, the outbound HTTP server has been removed from the cached
stack so that it is not included in gatewayed services.

Furthermore, to test this, this change splits `push_http_client` from
the `Outbound::push_http_endpoint` stack and `push_http_tcp_server` from
the `Inbound::push_http_server` stack. This allows us to write tests
that mock HTTP-level services.

Author: olix0r

Cargo.lock

@@ -859,6 +859,7 @@ dependencies = [
  "linkerd-app-inbound",
  "linkerd-app-outbound",
  "linkerd-app-test",
+ "linkerd-proxy-server-policy",
  "thiserror",
  "tokio",
  "tokio-test",

linkerd/app/gateway/Cargo.toml

@@ -18,6 +18,9 @@ tower = { version = "0.4", default-features = false }
 tracing = "0.1"
 
 [dev-dependencies]
+linkerd-app-inbound = { path = "../inbound", features = ["test-util"] }
+linkerd-app-outbound = { path = "../outbound", features = ["test-util"] }
+linkerd-proxy-server-policy = { path = "../../proxy/server-policy" }
 tokio = { version = "1", features = ["rt", "macros"] }
 tokio-test = "0.4"
 tower = { version = "0.4", default-features = false, features = ["util"] }

linkerd/app/gateway/src/http.rs

@@ -1,6 +1,18 @@
-use super::{server::Http, Gateway};
+use super::Gateway;
 use inbound::{GatewayAddr, GatewayDomainInvalid};
-use linkerd_app_core::{identity, io, profiles, proxy::http, svc, tls, transport::addrs::*, Error};
+use linkerd_app_core::{
+    identity,
+    metrics::ServerLabel,
+    profiles,
+    proxy::{
+        api_resolve::{ConcreteAddr, Metadata},
+        core::Resolve,
+        http,
+    },
+    svc, tls,
+    transport::addrs::*,
+    Error,
+};
 use linkerd_app_inbound as inbound;
 use linkerd_app_outbound as outbound;
 use std::{
@@ -34,34 +46,61 @@ pub struct Target<T = ()> {
 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>>
+    /// Wrap the provided outbound HTTP client with the inbound HTTP server,
+    /// inbound authorization, tagged-transport gateway routing, and the
+    /// outbound router.
+    pub fn http<T, N, R, NSvc>(
+        &self,
+        inner: N,
+        resolve: R,
+    ) -> svc::Stack<
+        svc::ArcNewService<
+            T,
+            impl svc::Service<
+                    http::Request<http::BoxBody>,
+                    Response = http::Response<http::BoxBody>,
+                    Error = Error,
+                    Future = impl Send,
+                > + Clone,
+        >,
+    >
     where
         // Target describing an inbound gateway connection.
         T: svc::Param<GatewayAddr>,
         T: svc::Param<OrigDstAddr>,
         T: svc::Param<Remote<ClientAddr>>,
+        T: svc::Param<ServerLabel>,
         T: svc::Param<tls::ConditionalServerTls>,
         T: svc::Param<tls::ClientId>,
         T: svc::Param<inbound::policy::AllowPolicy>,
-        T: svc::Param<profiles::LookupAddr>,
+        T: svc::Param<Option<profiles::Receiver>>,
+        T: svc::Param<http::Version>,
+        T: svc::Param<http::normalize_uri::DefaultAuthority>,
         T: Clone + Send + Sync + Unpin + 'static,
-        // Server-side socket.
-        I: io::AsyncRead + io::AsyncWrite + io::PeerAddr,
-        I: Send + Unpin + 'static,
+        // Endpoint resolution.
+        R: Resolve<ConcreteAddr, Endpoint = Metadata, Error = Error>,
         // HTTP outbound stack.
-        N: svc::NewService<Target, Service = NSvc>,
+        N: svc::NewService<
+            outbound::http::concrete::Endpoint<
+                outbound::http::logical::Concrete<outbound::http::Http>,
+            >,
+            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,
+        NSvc::Future: Send + Unpin + 'static,
     {
-        let http = svc::stack(inner)
+        let http = self
+            .outbound
+            .clone()
+            .with_stack(inner)
+            .push_http_cached(resolve)
+            .into_stack()
             // Discard `T` and its associated client-specific metadata.
             .push_map_target(Target::discard_parent)
             // Add headers to prevent loops.
@@ -77,29 +116,27 @@ impl Gateway {
             }))
             // 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(),
-                    })
-                },
-            )
+            .push_filter(|(_, parent): (_, 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: parent.param(),
+                    parent,
+                })
+            })
             // Authorize requests to the gateway.
             .push(self.inbound.authorize_http());