Make fallback HTTP-agnostic (#393)

The fallback module is currently HTTP-aware, as it must make response
body types uniform.

This change modifies fallback to instead require that both _service
types_ are identical. This is accomplished by boxing the services, using
a new http boxed service type.

This has the side-effect of dramatically improving compilation times for
the outbound proxy.

Author: olix0r

Cargo.lock

@@ -736,10 +736,7 @@ dependencies = [
 name = "linkerd2-fallback"
 version = "0.1.0"
 dependencies = [
- "bytes 0.4.11 (registry+https://github.com/rust-lang/crates.io-index)",
  "futures 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
- "http 0.1.16 (registry+https://github.com/rust-lang/crates.io-index)",
- "hyper 0.12.34 (registry+https://github.com/rust-lang/crates.io-index)",
  "linkerd2-error 0.1.0",
  "tower 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "tracing 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)",

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

@@ -1,5 +1,6 @@
 //! Tools for building a transparent TCP/HTTP proxy.
 
+pub use linkerd2_fallback as fallback;
 pub use linkerd2_proxy_api_resolve as api_resolve;
 pub use linkerd2_proxy_core as core;
 pub use linkerd2_proxy_detect as detect;

linkerd/app/core/src/svc.rs

@@ -1,4 +1,5 @@
-use crate::proxy::{buffer, pending};
+use crate::proxy::{buffer, http, pending};
+use crate::Error;
 pub use linkerd2_router::Make;
 pub use linkerd2_stack::{self as stack, layer, map_target, Layer, LayerExt, Shared};
 pub use linkerd2_timeout::stack as timeout;
@@ -53,6 +54,14 @@ impl<L> Layers<L> {
     pub fn push_spawn_ready(self) -> Layers<Pair<L, SpawnReadyLayer>> {
         self.push(SpawnReadyLayer::new())
     }
+
+    pub fn boxed<A, B>(self) -> Layers<Pair<L, http::boxed::Layer<A, B>>>
+    where
+        A: 'static,
+        B: hyper::body::Payload<Data = http::boxed::Data, Error = Error> + 'static,
+    {
+        self.push(http::boxed::Layer::new())
+    }
 }
 
 impl<M, L: Layer<M>> Layer<M> for Layers<L> {
@@ -104,6 +113,15 @@ impl<S> Stack<S> {
         self.push(TimeoutLayer::new(timeout))
     }
 
+    pub fn boxed<T, A, B>(self) -> Stack<http::boxed::Make<S, A, B>>
+    where
+        A: 'static,
+        S: tower::MakeService<T, http::Request<A>, Response = http::Response<B>>,
+        B: hyper::body::Payload<Data = http::boxed::Data, Error = Error> + 'static,
+    {
+        self.push(http::boxed::Layer::new())
+    }
+
     /// Validates that this stack serves T-typed targets.
     pub fn makes<T>(self) -> Self
     where

linkerd/app/outbound/src/lib.rs

@@ -15,15 +15,7 @@ use linkerd2_app_core::{
     http_request_l5d_override_dst_addr, http_request_orig_dst_addr,
     opencensus::proto::trace::v1 as oc,
     proxy::{
-        self,
-        core::resolve::Resolve,
-        discover,
-        http::{
-            balance, canonicalize, client, fallback, header_from_target, insert,
-            metrics as http_metrics, normalize_uri, profiles, retry, settings, strip_header,
-        },
-        identity,
-        resolve::map_endpoint,
+        self, core::resolve::Resolve, discover, fallback, http, identity, resolve::map_endpoint,
         tap, tcp, Server,
     },
     reconnect, router, serve,
@@ -132,15 +124,15 @@ impl<A: OrigDstAddr> Config<A> {
             // Instantiates an HTTP client for for a `client::Config`
             let client_stack = connect_stack
                 .clone()
-                .push(client::layer(connect.h2_settings))
+                .push(http::client::layer(connect.h2_settings))
                 .push(reconnect::layer({
                     let backoff = connect.backoff.clone();
                     move |_| Ok(backoff.stream())
                 }))
                 .push(trace_context::layer(span_sink.clone().map(|span_sink| {
                     SpanConverter::client(span_sink, trace_labels())
                 })))
-                .push(normalize_uri::layer());
+                .push(http::normalize_uri::layer());
 
             // A per-`outbound::Endpoint` stack that:
             //
@@ -156,21 +148,22 @@ impl<A: OrigDstAddr> Config<A> {
             //    the server, before we apply our own.
             let endpoint_stack = client_stack
                 .serves::<Endpoint>()
-                .push(strip_header::response::layer(L5D_REMOTE_IP))
-                .push(strip_header::response::layer(L5D_SERVER_ID))
-                .push(strip_header::request::layer(L5D_REQUIRE_ID))
+                .push(http::strip_header::response::layer(L5D_REMOTE_IP))
+                .push(http::strip_header::response::layer(L5D_SERVER_ID))
+                .push(http::strip_header::request::layer(L5D_REQUIRE_ID))
                 // disabled due to information leagkage
                 //.push(add_remote_ip_on_rsp::layer())
                 //.push(add_server_id_on_rsp::layer())
                 .push(orig_proto_upgrade::layer())
                 .push(tap_layer.clone())
-                .push(http_metrics::layer::<_, classify::Response>(
+                .push(http::metrics::layer::<_, classify::Response>(
                     metrics.http_endpoint,
                 ))
                 .push(require_identity_on_endpoint::layer())
                 .push(trace::layer(|endpoint: &Endpoint| {
                     info_span!("endpoint", peer.addr = %endpoint.addr, peer.id = ?endpoint.identity)
-                }));
+                }))
+                .serves::<Endpoint>();
 
             // A per-`dst::Route` layer that uses profile data to configure
             // a per-route layer.
@@ -184,13 +177,13 @@ impl<A: OrigDstAddr> Config<A> {
             // 3. Retries are optionally enabled depending on if the route
             //    is retryable.
             let dst_route_layer = svc::layers()
-                .push(insert::target::layer())
-                .push(http_metrics::layer::<_, classify::Response>(
+                .push(http::insert::target::layer())
+                .push(http::metrics::layer::<_, classify::Response>(
                     metrics.http_route_retry.clone(),
                 ))
-                .push(retry::layer(metrics.http_route_retry))
-                .push(proxy::http::timeout::layer())
-                .push(http_metrics::layer::<_, classify::Response>(
+                .push(http::retry::layer(metrics.http_route_retry))
+                .push(http::timeout::layer())
+                .push(http::metrics::layer::<_, classify::Response>(
                     metrics.http_route,
                 ))
                 .push(classify::layer())
@@ -215,16 +208,19 @@ impl<A: OrigDstAddr> Config<A> {
                 .push_spawn_ready()
                 .push(discover::Layer::new(
                     DISCOVER_UPDATE_BUFFER_CAPACITY,
-                    map_endpoint::Resolve::new(endpoint::FromMetadata, resolve),
+                    map_endpoint::Resolve::new(endpoint::FromMetadata, resolve.clone()),
                 ))
-                .push(balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY));
+                .push(http::balance::layer(EWMA_DEFAULT_RTT, EWMA_DECAY));
 
             // If the balancer fails to be created, i.e., because it is unresolvable,
             // fall back to using a router that dispatches request to the
             // application-selected original destination.
             let distributor = endpoint_stack
-                .push(fallback::layer(balancer_layer, orig_dst_router_layer))
-                .serves::<DstAddr>()
+                .serves::<Endpoint>()
+                .push(fallback::layer(
+                    balancer_layer.boxed(),
+                    orig_dst_router_layer.boxed(),
+                ))
                 .push(trace::layer(
                     |dst: &DstAddr| info_span!("concrete", dst.concrete = %dst.dst_concrete()),
                 ));
@@ -237,9 +233,14 @@ impl<A: OrigDstAddr> Config<A> {
             // 3. Creates a load balancer , configured by resolving the
             //   `DstAddr` with a resolver.
             let dst_stack = distributor
+                .serves::<DstAddr>()
                 .push_buffer_pending(buffer.max_in_flight, DispatchDeadline::extract)
-                .push(profiles::router::layer(profiles_client, dst_route_layer))
-                .push(header_from_target::layer(CANONICAL_DST_HEADER));
+                .makes::<DstAddr>()
+                .push(http::profiles::router::layer(
+                    profiles_client,
+                    dst_route_layer,
+                ))
+                .push(http::header_from_target::layer(CANONICAL_DST_HEADER));
 
             // Routes request using the `DstAddr` extension.
             //
@@ -254,7 +255,7 @@ impl<A: OrigDstAddr> Config<A> {
                     router::Config::new(router_capacity, router_max_idle_age),
                     |req: &http::Request<_>| {
                         req.extensions().get::<Addr>().cloned().map(|addr| {
-                            DstAddr::outbound(addr, settings::Settings::from_request(req))
+                            DstAddr::outbound(addr, http::settings::Settings::from_request(req))
                         })
                     },
                 ))
@@ -264,8 +265,9 @@ impl<A: OrigDstAddr> Config<A> {
             // Canonicalizes the request-specified `Addr` via DNS, and
             // annotates each request with a refined `Addr` so that it may be
             // routed by the dst_router.
-            let addr_stack = svc::stack(svc::Shared::new(dst_router))
-                .push(canonicalize::layer(dns_resolver, canonicalize_timeout));
+            let addr_stack = svc::stack(svc::Shared::new(dst_router)).push(
+                http::canonicalize::layer(dns_resolver, canonicalize_timeout),
+            );
 
             // Routes requests to an `Addr`:
             //
@@ -283,9 +285,9 @@ impl<A: OrigDstAddr> Config<A> {
             // 5. Finally, if the tls::accept::Meta had an SO_ORIGINAL_DST, this TCP
             // address is used.
             let addr_router = addr_stack
-                .push(strip_header::request::layer(L5D_CLIENT_ID))
-                .push(strip_header::request::layer(DST_OVERRIDE_HEADER))
-                .push(insert::target::layer())
+                .push(http::strip_header::request::layer(L5D_CLIENT_ID))
+                .push(http::strip_header::request::layer(DST_OVERRIDE_HEADER))
+                .push(http::insert::target::layer())
                 .push(trace::layer(|addr: &Addr| info_span!("addr", %addr)))
                 .push_buffer_pending(buffer.max_in_flight, DispatchDeadline::extract)
                 .push(router::Layer::new(
@@ -315,17 +317,17 @@ impl<A: OrigDstAddr> Config<A> {
             // shared `addr_router`. The `tls::accept::Meta` is stored in the request's
             // extensions so that it can be used by the `addr_router`.
             let server_stack = svc::stack(svc::Shared::new(admission_control))
-                .push(insert::layer(move || {
+                .push(http::insert::layer(move || {
                     DispatchDeadline::after(buffer.dispatch_timeout)
                 }))
-                .push(insert::target::layer())
+                .push(http::insert::target::layer())
                 .push(errors::layer())
                 .push(trace::layer(
                     |src: &tls::accept::Meta| info_span!("source", target.addr = %src.addrs.target_addr()),
                 ))
-                .push(trace_context::layer(
-                span_sink.map(|span_sink| SpanConverter::server(span_sink, trace_labels())),
-            ))
+                .push(trace_context::layer(span_sink.map(|span_sink| {
+                    SpanConverter::server(span_sink, trace_labels())
+                })))
                 .push(metrics.http_handle_time.layer());
 
             let forward_tcp = tcp::Forward::new(

linkerd/fallback/Cargo.toml

@@ -6,10 +6,7 @@ edition = "2018"
 publish = false
 
 [dependencies]
-bytes = "0.4"
 futures = "0.1"
-http = "0.1"
-hyper = "0.12"
 linkerd2-error = { path = "../error" }
 tower= "0.1"
 tracing = "0.1"