outbound: Mock the original destination address in ingress-mode (#942)

The ingress-mode outbound proxy currently does not work as expected:
resolutions are performed for each unique endpoint so that a service
with 10 endpoints would manifest as 10 distinct balancers (with 20
resolutions).

While we should ultimately rework the outbound stack to have target
types that prevent this problem (i.e. by removing the original
destination address from the Logical target type), this change
implements a hack to avoid the observed cardinality issues: In
ingress-mode, the original destination address is now mocked out with an
IP address of 0.0.0.0 so that only a single cache entry is created per
logical destination.

Author: olix0r

linkerd/app/outbound/src/ingress.rs

@@ -3,7 +3,7 @@ use linkerd_app_core::{
     config::{ProxyConfig, ServerConfig},
     detect, discovery_rejected, drain, errors, http_request_l5d_override_dst_addr, http_tracing,
     io, profiles, svc, tls,
-    transport::{self, listen},
+    transport::{self, listen, OrigDstAddr},
     Addr, AddrMatch, Error,
 };
 use std::convert::TryFrom;
@@ -104,8 +104,10 @@ impl Outbound<()> {
             // Note that the router service is always ready, so the `FailFast` layer
             // need not use a `SpawnReady` to drive the service to ready.
             .push(svc::NewRouter::layer(TargetPerRequest::accept))
+            .push(http::NewNormalizeUri::layer())
             .push_on_response(
                 svc::layers()
+                    .push(http::MarkAbsoluteForm::layer())
                     .push(svc::ConcurrencyLimit::layer(max_in_flight_requests))
                     .push(svc::FailFast::layer("HTTP Server", dispatch_timeout))
                     .push(self.runtime.metrics.http_errors.clone())
@@ -117,8 +119,6 @@ impl Outbound<()> {
                     .push(http::BoxResponse::layer()),
             )
             .check_new_service::<http::Accept, http::Request<_>>()
-            .push(http::NewNormalizeUri::layer())
-            .check_new_service::<http::Accept, http::Request<_>>()
             .instrument(|a: &http::Accept| debug_span!("http", v = %a.protocol))
             .check_new_service::<http::Accept, http::Request<_>>()
             .push(http::NewServeHttp::layer(
@@ -150,7 +150,7 @@ struct AllowHttpProfile(AddrMatch);
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
 struct Target {
     dst: Addr,
-    accept: http::Accept,
+    version: http::Version,
 }
 
 #[derive(Clone)]
@@ -173,11 +173,18 @@ impl svc::stack::Predicate<Target> for AllowHttpProfile {
 // === impl Target ===
 
 impl From<(Option<profiles::Receiver>, Target)> for http::Logical {
-    fn from((p, Target { accept, .. }): (Option<profiles::Receiver>, Target)) -> Self {
+    fn from((profile, Target { dst, version }): (Option<profiles::Receiver>, Target)) -> Self {
+        // XXX This is a hack to fix caching when an dst-override is set.
+        let orig_dst = if let Some(a) = dst.socket_addr() {
+            OrigDstAddr(a)
+        } else {
+            OrigDstAddr(([0, 0, 0, 0], dst.port()).into())
+        };
+
         Self {
-            profile: p,
-            orig_dst: accept.orig_dst,
-            protocol: accept.protocol,
+            orig_dst,
+            profile,
+            protocol: version,
         }
     }
 }
@@ -194,10 +201,11 @@ impl<B> svc::stack::RecognizeRoute<http::Request<B>> for TargetPerRequest {
     type Key = Target;
 
     fn recognize(&self, req: &http::Request<B>) -> Result<Self::Key, Error> {
+        let dst =
+            http_request_l5d_override_dst_addr(req).unwrap_or_else(|_| self.0.orig_dst.0.into());
         Ok(Target {
-            accept: self.0,
-            dst: http_request_l5d_override_dst_addr(req)
-                .unwrap_or_else(|_| self.0.orig_dst.0.into()),
+            dst,
+            version: self.0.protocol,
         })
     }
 }

linkerd/proxy/http/src/lib.rs

@@ -29,7 +29,7 @@ pub use self::{
     detect::DetectHttp,
     glue::{HyperServerSvc, UpgradeBody},
     header_from_target::NewHeaderFromTarget,
-    normalize_uri::NewNormalizeUri,
+    normalize_uri::{MarkAbsoluteForm, NewNormalizeUri},
     override_authority::{CanOverrideAuthority, NewOverrideAuthority},
     retain::Retain,
     server::NewServeHttp,