outbound: Use the OrigDstAddr type in stack targets (#936)

The outbound stack is ambiguous about the type of orig_dst/target
address it uses.

This change modifies the outbound stack types to use the `OrigDstAddr`
type. The outbound stacks now error when an `OrigDstAddr` type is not
present.

Author: olix0r

linkerd/app/gateway/src/gateway.rs

@@ -6,7 +6,9 @@ use linkerd_app_core::{
     profiles,
     proxy::http,
     svc::{self, layer},
-    tls, Error, NameAddr,
+    tls,
+    transport::OrigDstAddr,
+    Error, NameAddr,
 };
 use linkerd_app_outbound as outbound;
 use std::{
@@ -71,7 +73,7 @@ where
         let svc = self.outbound.new_service(outbound::http::Logical {
             profile,
             protocol: http.version,
-            orig_dst: ([0, 0, 0, 0], dst.port()).into(),
+            orig_dst: OrigDstAddr(([0, 0, 0, 0], dst.port()).into()),
         });
 
         Gateway::new(svc, http.target, local_id)

linkerd/app/gateway/src/lib.rs

@@ -11,6 +11,7 @@ use linkerd_app_core::{
     proxy::{api_resolve::Metadata, core::Resolve, http},
     svc::{self, Param},
     tls,
+    transport::OrigDstAddr,
     transport_header::SessionProtocol,
     Error, NameAddr, NameMatch, Never,
 };
@@ -120,7 +121,7 @@ where
         .push_request_filter(|(p, _): (Option<profiles::Receiver>, _)| match p {
             Some(rx) if rx.borrow().name.is_some() => Ok(outbound::tcp::Logical {
                 profile: Some(rx),
-                orig_dst: std::net::SocketAddr::from(([0, 0, 0, 0], 0)),
+                orig_dst: OrigDstAddr(std::net::SocketAddr::from(([0, 0, 0, 0], 0))),
                 protocol: (),
             }),
             _ => Err(discovery_rejected()),

linkerd/app/outbound/src/discover.rs

@@ -4,6 +4,7 @@ use linkerd_app_core::{
     transport::{listen, metrics::SensorIo},
     Error, IpMatch,
 };
+use std::convert::TryFrom;
 
 impl<N> Outbound<N> {
     /// Discovers the profile for a TCP endpoint.
@@ -58,7 +59,7 @@ impl<N> Outbound<N> {
             .push(rt.metrics.transport.layer_accept())
             .push_cache(config.proxy.cache_max_idle_age)
             .check_new_service::<tcp::Accept, I>()
-            .push_map_target(tcp::Accept::from)
+            .push_request_filter(tcp::Accept::try_from)
             .check_new_service::<listen::Addrs, I>();
 
         Outbound {
@@ -78,8 +79,8 @@ impl svc::stack::Predicate<tcp::Accept> for AllowProfile {
     type Request = profiles::LogicalAddr;
 
     fn check(&mut self, a: tcp::Accept) -> Result<profiles::LogicalAddr, Error> {
-        if self.0.matches(a.orig_dst.ip()) {
-            Ok(profiles::LogicalAddr(a.orig_dst.into()))
+        if self.0.matches(a.orig_dst.0.ip()) {
+            Ok(profiles::LogicalAddr(a.orig_dst.0.into()))
         } else {
             Err(discovery_rejected().into())
         }

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

@@ -72,7 +72,7 @@ impl Param<normalize_uri::DefaultAuthority> for Logical {
         if let Some(p) = self.profile.as_ref() {
             if let Some(n) = p.borrow().name.as_ref() {
                 return normalize_uri::DefaultAuthority(Some(
-                    uri::Authority::from_str(&format!("{}:{}", n, self.orig_dst.port()))
+                    uri::Authority::from_str(&format!("{}:{}", n, self.orig_dst.0.port()))
                         .expect("Address must be a valid authority"),
                 ));
             }

linkerd/app/outbound/src/ingress.rs

@@ -6,6 +6,7 @@ use linkerd_app_core::{
     transport::{self, listen},
     Addr, AddrMatch, Error,
 };
+use std::convert::TryFrom;
 use tracing::{debug_span, info_span};
 
 impl Outbound<()> {
@@ -132,7 +133,7 @@ impl Outbound<()> {
             ))
             .check_new_service::<tcp::Accept, transport::metrics::SensorIo<I>>()
             .push(self.runtime.metrics.transport.layer_accept())
-            .push_map_target(tcp::Accept::from)
+            .push_request_filter(tcp::Accept::try_from)
             .check_new_service::<listen::Addrs, I>()
             // Boxing is necessary purely to limit the link-time overhead of
             // having enormous types.
@@ -193,7 +194,8 @@ impl<B> svc::stack::RecognizeRoute<http::Request<B>> for TargetPerRequest {
     fn recognize(&self, req: &http::Request<B>) -> Result<Self::Key, Error> {
         Ok(Target {
             accept: self.0,
-            dst: http_request_l5d_override_dst_addr(req).unwrap_or_else(|_| self.0.orig_dst.into()),
+            dst: http_request_l5d_override_dst_addr(req)
+                .unwrap_or_else(|_| self.0.orig_dst.0.into()),
         })
     }
 }

linkerd/app/outbound/src/target.rs

@@ -6,7 +6,7 @@ use linkerd_app_core::{
     },
     svc::{self, Param},
     tls,
-    transport::{self, Remote, ServerAddr},
+    transport::{self, OrigDstAddr, Remote, ServerAddr},
     transport_header, Addr, Conditional, Error,
 };
 use std::net::SocketAddr;
@@ -16,13 +16,13 @@ pub struct EndpointFromMetadata;
 
 #[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
 pub struct Accept<P> {
-    pub orig_dst: SocketAddr,
+    pub orig_dst: OrigDstAddr,
     pub protocol: P,
 }
 
 #[derive(Clone)]
 pub struct Logical<P> {
-    pub orig_dst: SocketAddr,
+    pub orig_dst: OrigDstAddr,
     pub profile: Option<profiles::Receiver>,
     pub protocol: P,
 }
@@ -44,22 +44,14 @@ pub struct Endpoint<P> {
 
 // === impl Accept ===
 
-impl<P> Param<SocketAddr> for Accept<P> {
-    fn param(&self) -> SocketAddr {
-        self.orig_dst
-    }
-}
-
-impl<P> Param<Addr> for Accept<P> {
-    fn param(&self) -> Addr {
-        self.orig_dst.into()
-    }
-}
-
 impl<P> Param<transport::labels::Key> for Accept<P> {
     fn param(&self) -> transport::labels::Key {
         const NO_TLS: tls::ConditionalServerTls = Conditional::None(tls::NoServerTls::Loopback);
-        transport::labels::Key::accept(transport::labels::Direction::Out, NO_TLS, self.orig_dst)
+        transport::labels::Key::accept(
+            transport::labels::Direction::Out,
+            NO_TLS,
+            self.orig_dst.into(),
+        )
     }
 }
 
@@ -89,13 +81,6 @@ impl<P> Param<Option<profiles::Receiver>> for Logical<P> {
     }
 }
 
-/// Used to determine whether detection should be skipped.
-impl<P> Param<SocketAddr> for Logical<P> {
-    fn param(&self) -> SocketAddr {
-        self.orig_dst
-    }
-}
-
 /// Used for default traffic split
 impl<P> Param<profiles::LogicalAddr> for Logical<P> {
     fn param(&self) -> profiles::LogicalAddr {
@@ -108,8 +93,8 @@ impl<P> Logical<P> {
         self.profile
             .as_ref()
             .and_then(|p| p.borrow().name.clone())
-            .map(|n| Addr::from((n, self.orig_dst.port())))
-            .unwrap_or_else(|| self.orig_dst.into())
+            .map(|n| Addr::from((n, self.orig_dst.0.port())))
+            .unwrap_or_else(|| self.orig_dst.0.into())
     }
 }
 
@@ -194,7 +179,7 @@ impl<P> Endpoint<P> {
             .and_then(|p| p.borrow().endpoint.clone())
         {
             None => Self {
-                addr: Remote(ServerAddr(logical.orig_dst)),
+                addr: Remote(ServerAddr(logical.orig_dst.into())),
                 metadata: Metadata::default(),
                 tls: Conditional::None(reason),
                 logical_addr: logical.addr(),

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

@@ -6,22 +6,40 @@ mod tests;
 
 use crate::target;
 pub use linkerd_app_core::proxy::tcp::Forward;
-use linkerd_app_core::{svc::Param, transport::listen, transport_header::SessionProtocol};
+use linkerd_app_core::{
+    svc::Param,
+    transport::{listen, OrigDstAddr},
+    transport_header::SessionProtocol,
+};
 
 pub type Accept = target::Accept<()>;
 pub type Logical = target::Logical<()>;
 pub type Concrete = target::Concrete<()>;
 pub type Endpoint = target::Endpoint<()>;
 
-impl From<listen::Addrs> for Accept {
-    fn from(addrs: listen::Addrs) -> Self {
+impl From<OrigDstAddr> for Accept {
+    fn from(orig_dst: OrigDstAddr) -> Self {
         Self {
-            orig_dst: addrs.target_addr(),
+            orig_dst,
             protocol: (),
         }
     }
 }
 
+impl std::convert::TryFrom<listen::Addrs> for Accept {
+    type Error = std::io::Error;
+
+    fn try_from(t: listen::Addrs) -> Result<Self, Self::Error> {
+        match t.orig_dst() {
+            Some(addr) => Ok(Self::from(addr)),
+            None => Err(std::io::Error::new(
+                std::io::ErrorKind::NotFound,
+                "No SO_ORIGINAL_DST address found",
+            )),
+        }
+    }
+}
+
 impl<P> From<(P, Accept)> for target::Accept<P> {
     fn from((protocol, Accept { orig_dst, .. }): (P, Accept)) -> Self {
         Self { orig_dst, protocol }

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

@@ -38,7 +38,7 @@ async fn plaintext_tcp() {
     // address to resolve, etc.
     let target_addr = SocketAddr::new([0, 0, 0, 0].into(), 666);
     let logical = Logical {
-        orig_dst: target_addr,
+        orig_dst: OrigDstAddr(target_addr),
         profile: Some(profile::only_default()),
         protocol: (),
     };
@@ -81,14 +81,16 @@ async fn plaintext_tcp() {
 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: SocketAddr::new([0, 0, 0, 0].into(), 5550),
+        orig_dst: OrigDstAddr(tls_addr),
         profile: Some(profile::only_default()),
         protocol: (),
     };
 
+    let plain_addr = SocketAddr::new([0, 0, 0, 0].into(), 5551);
     let plain = Logical {
-        orig_dst: SocketAddr::new([0, 0, 0, 0].into(), 5551),
+        orig_dst: OrigDstAddr(plain_addr),
         profile: Some(profile::only_default()),
         protocol: (),
     };
@@ -103,12 +105,12 @@ async fn tls_when_hinted() {
     // Build a mock "connector" that returns the upstream "server" IO.
     let connect = support::connect()
         // The plaintext endpoint should use plaintext...
-        .endpoint_fn(plain.orig_dst, move |endpoint: Endpoint| {
+        .endpoint_fn(plain_addr, move |endpoint: Endpoint| {
             assert!(endpoint.tls.is_none());
             let io = tls_srv_io.build();
             Ok(io)
         })
-        .endpoint_fn(tls.orig_dst, move |endpoint: Endpoint| {
+        .endpoint_fn(tls_addr, move |endpoint: Endpoint| {
             assert_eq!(endpoint.tls, Conditional::Some(id_name2.clone().into()));
             let io = srv_io.build();
             Ok(io)
@@ -117,10 +119,10 @@ async fn tls_when_hinted() {
     // Configure the mock destination resolver to just give us a single endpoint
     // for the target, which always exists and has no metadata.
     let resolver = support::resolver()
-        .endpoint_exists(plain.orig_dst, plain.orig_dst, Default::default())
+        .endpoint_exists(plain_addr, plain_addr, Default::default())
         .endpoint_exists(
-            tls.orig_dst,
-            tls.orig_dst,
+            tls_addr,
+            tls_addr,
             support::resolver::Metadata::new(
                 Default::default(),
                 support::resolver::ProtocolHint::Unknown,
@@ -136,15 +138,15 @@ async fn tls_when_hinted() {
 
     // Build the outbound TCP balancer stack.
     let (rt, _) = runtime();
-    let plain = Outbound::new(default_config(plain.orig_dst), rt.clone())
+    let plain = Outbound::new(default_config(plain_addr), rt.clone())
         .with_stack(connect.clone())
         .push_tcp_logical(resolver.clone())
         .into_inner()
         .new_service(plain)
         .oneshot(client_io.build())
         .err_into::<Error>();
 
-    let tls = Outbound::new(default_config(tls.orig_dst), rt)
+    let tls = Outbound::new(default_config(tls_addr), rt)
         .with_stack(connect)
         .push_tcp_logical(resolver)
         .into_inner()