Fix caching in ingress mode (#965)

When the proxy is in `ingress` mode, requests with different
`l5d-dst-override` headers but the same address share the same key in the
cache. This causes consecutive requests to the same address to all go to the
value of the first `l5d-dst-override` header, ignoring the header values of
later requests—unless the services are evicted from the cache.

This occurs because each key into the cache is calculated based off
`LogicalAddr`'s `orig_dst` and `protocol` fields. When the proxy is in
`ingress` mode, the value of `orig_dst` is overridden to `0.0.0.0:port` This
results in an easy key conflict: two different requests with the same port
and protocol with result in the same key.

To fix this, the `logical_addr` field is added to `LogicalAddr` and is used
as a third field in calculating the cache key. `logical_addr` is the value of
the `l5d-dst-override` header which means requests to the same address but
with different header values will result in different keys.

Fixes linkerd/linkerd2#5975

Signed-off-by: Kevin Leimkuhler <[email protected]>

Author: kleimkuhler

linkerd/app/gateway/src/gateway.rs

@@ -61,8 +61,8 @@ where
             None => return Gateway::NoIdentity,
         };
 
-        let dst = match profile.as_ref().and_then(|p| p.borrow().addr.clone()) {
-            Some(profiles::LogicalAddr(addr)) => addr,
+        let addr = match profile.as_ref().and_then(|p| p.borrow().addr.clone()) {
+            Some(addr) => addr,
             None => return Gateway::BadDomain(http.target.name().clone()),
         };
 
@@ -73,7 +73,8 @@ where
         let svc = self.outbound.new_service(outbound::http::Logical {
             profile,
             protocol: http.version,
-            orig_dst: OrigDstAddr(([0, 0, 0, 0], dst.port()).into()),
+            orig_dst: OrigDstAddr(([0, 0, 0, 0], addr.0.port()).into()),
+            logical_addr: Some(addr),
         });
 
         Gateway::new(svc, http.target, local_id)

linkerd/app/gateway/src/lib.rs

@@ -116,11 +116,15 @@ where
         .push_tcp_logical(resolve.clone())
         .into_stack()
         .push_request_filter(|(p, _): (Option<profiles::Receiver>, _)| match p {
-            Some(rx) if rx.borrow().addr.is_some() => Ok(outbound::tcp::Logical {
-                profile: Some(rx),
-                orig_dst: OrigDstAddr(std::net::SocketAddr::from(([0, 0, 0, 0], 0))),
-                protocol: (),
-            }),
+            Some(rx) if rx.borrow().addr.is_some() => {
+                let logical_addr = rx.borrow().addr.clone();
+                Ok(outbound::tcp::Logical {
+                    profile: Some(rx),
+                    orig_dst: OrigDstAddr(std::net::SocketAddr::from(([0, 0, 0, 0], 0))),
+                    protocol: (),
+                    logical_addr,
+                })
+            }
             _ => Err(discovery_rejected()),
         })
         .push(profiles::discover::layer(profiles.clone(), {

linkerd/app/outbound/src/http/mod.rs

@@ -65,6 +65,7 @@ impl From<(Version, tcp::Logical)> for Logical {
             protocol,
             orig_dst: logical.orig_dst,
             profile: logical.profile,
+            logical_addr: logical.logical_addr,
         }
     }
 }

linkerd/app/outbound/src/ingress.rs

@@ -183,11 +183,13 @@ impl From<(Option<profiles::Receiver>, Target)> for http::Logical {
         } else {
             OrigDstAddr(([0, 0, 0, 0], dst.port()).into())
         };
+        let logical_addr = profile.as_ref().and_then(|p| p.borrow().addr.clone());
 
         Self {
             orig_dst,
             profile,
             protocol: version,
+            logical_addr,
         }
     }
 }

linkerd/app/outbound/src/target.rs

@@ -24,6 +24,7 @@ pub struct Accept<P> {
 #[derive(Clone)]
 pub struct Logical<P> {
     pub orig_dst: OrigDstAddr,
+    pub logical_addr: Option<LogicalAddr>,
     pub profile: Option<profiles::Receiver>,
     pub protocol: P,
 }
@@ -72,10 +73,12 @@ impl<P> From<(Option<profiles::Receiver>, Accept<P>)> for Logical<P> {
             },
         ): (Option<profiles::Receiver>, Accept<P>),
     ) -> Self {
+        let logical_addr = profile.as_ref().and_then(|p| p.borrow().addr.clone());
         Self {
             profile,
             orig_dst,
             protocol,
+            logical_addr,
         }
     }
 }
@@ -96,17 +99,18 @@ impl<P> Param<profiles::LookupAddr> for Logical<P> {
 
 impl<P> Logical<P> {
     pub fn addr(&self) -> Addr {
-        self.profile
+        self.logical_addr
             .as_ref()
-            .and_then(|p| p.borrow().addr.clone())
-            .map(|LogicalAddr(a)| Addr::from(a))
+            .map(|LogicalAddr(a)| Addr::from(a.clone()))
             .unwrap_or_else(|| self.orig_dst.0.into())
     }
 }
 
 impl<P: PartialEq> PartialEq<Logical<P>> for Logical<P> {
     fn eq(&self, other: &Logical<P>) -> bool {
-        self.orig_dst == other.orig_dst && self.protocol == other.protocol
+        self.orig_dst == other.orig_dst
+            && self.logical_addr == other.logical_addr
+            && self.protocol == other.protocol
     }
 }
 
@@ -115,6 +119,7 @@ impl<P: Eq> Eq for Logical<P> {}
 impl<P: std::hash::Hash> std::hash::Hash for Logical<P> {
     fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
         self.orig_dst.hash(state);
+        self.logical_addr.hash(state);
         self.protocol.hash(state);
     }
 }
@@ -135,6 +140,7 @@ impl<P: std::fmt::Debug> std::fmt::Debug for Logical<P> {
                     }
                 ),
             )
+            .field("logical_addr", &self.logical_addr)
             .finish()
     }
 }

linkerd/app/outbound/src/tcp/tests.rs

@@ -10,7 +10,9 @@ use crate::{
     Config, Outbound,
 };
 use linkerd_app_core::{
-    io, svc,
+    io,
+    profiles::Receiver,
+    svc,
     svc::NewService,
     tls,
     transport::{addrs::*, listen, orig_dst},
@@ -37,11 +39,8 @@ async fn plaintext_tcp() {
     // ports or anything. These will just be used so that the proxy has a socket
     // address to resolve, etc.
     let target_addr = SocketAddr::new([0, 0, 0, 0].into(), 666);
-    let logical = Logical {
-        orig_dst: OrigDstAddr(target_addr),
-        profile: Some(profile::only_default()),
-        protocol: (),
-    };
+    let profile_recv = profile::only_default();
+    let logical = build_logical(target_addr, profile_recv);
 
     // Configure mock IO for the upstream "server". It will read "hello" and
     // then write "world".
@@ -80,18 +79,12 @@ async fn tls_when_hinted() {
     let _trace = support::trace_init();
 
     let tls_addr = SocketAddr::new([0, 0, 0, 0].into(), 5550);
-    let tls = Logical {
-        orig_dst: OrigDstAddr(tls_addr),
-        profile: Some(profile::only_with_addr("tls:5550")),
-        protocol: (),
-    };
+    let tls_recv = profile::only_with_addr("tls:5550");
+    let tls = build_logical(tls_addr, tls_recv);
 
     let plain_addr = SocketAddr::new([0, 0, 0, 0].into(), 5551);
-    let plain = Logical {
-        orig_dst: OrigDstAddr(plain_addr),
-        profile: Some(profile::only_with_addr("plain:5551")),
-        protocol: (),
-    };
+    let plain_recv = profile::only_with_addr("plain:5551");
+    let plain = build_logical(plain_addr, plain_recv);
 
     let id_name = tls::ServerId::from_str("foo.ns1.serviceaccount.identity.linkerd.cluster.local")
         .expect("hostname is valid");
@@ -822,3 +815,13 @@ where
     }
     .instrument(span)
 }
+
+fn build_logical(addr: SocketAddr, profile_recv: Receiver) -> Logical {
+    let logical_addr = profile_recv.borrow().addr.clone();
+    Logical {
+        orig_dst: OrigDstAddr(addr),
+        profile: Some(profile_recv),
+        protocol: (),
+        logical_addr,
+    }
+}