Update profile response to include a logical address (#954)

Profile responses optionally include a service name, but we only ever
use it in the context of an address (i.e. with a port).

This change introduces a `profiles::LookupAddr` param type that is used
as input into the profile client -- the lookup address may hold either a
named or numbered address. Service profile responses now include a
`LogicalAddr` that must be a named address.

This sets up for requiring that concrete addresses are named.

Author: olix0r

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().name.clone()) {
-            Some(name) => NameAddr::from((name, http.target.port())),
+        let dst = match profile.as_ref().and_then(|p| p.borrow().addr.clone()) {
+            Some(profiles::LogicalAddr(addr)) => addr,
             None => return Gateway::BadDomain(http.target.name().clone()),
         };
 

linkerd/app/gateway/src/lib.rs

@@ -84,7 +84,7 @@ where
     O::Response:
         io::AsyncRead + io::AsyncWrite + tls::HasNegotiatedProtocol + Send + Unpin + 'static,
     O::Future: Send + Unpin + 'static,
-    P: profiles::GetProfile<profiles::LogicalAddr> + Clone + Send + Sync + Unpin + 'static,
+    P: profiles::GetProfile<profiles::LookupAddr> + Clone + Send + Sync + Unpin + 'static,
     P::Future: Send + 'static,
     P::Error: Send,
     R: Clone + Send + Sync + Unpin + 'static,
@@ -116,7 +116,7 @@ where
         .push_tcp_logical(resolve.clone())
         .into_stack()
         .push_request_filter(|(p, _): (Option<profiles::Receiver>, _)| match p {
-            Some(rx) if rx.borrow().name.is_some() => Ok(outbound::tcp::Logical {
+            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: (),
@@ -127,7 +127,7 @@ where
             let allow = allow_discovery.clone();
             move |addr: NameAddr| {
                 if allow.matches(addr.name()) {
-                    Ok(profiles::LogicalAddr(addr.into()))
+                    Ok(profiles::LookupAddr(addr.into()))
                 } else {
                     Err(RefusedNotResolved(addr))
                 }
@@ -162,7 +162,7 @@ where
         .push(NewGateway::layer(local_id))
         .push(profiles::discover::layer(profiles, move |t: HttpTarget| {
             if allow_discovery.matches(t.target.name()) {
-                Ok(profiles::LogicalAddr(t.target.into()))
+                Ok(profiles::LookupAddr(t.target.into()))
             } else {
                 Err(RefusedNotResolved(t.target))
             }

linkerd/app/gateway/src/tests.rs

@@ -104,7 +104,7 @@ impl Test {
         let allow = NameMatch::new(Some(dns::Suffix::from_str(suffix).unwrap()));
         let profile = if allow.matches(target.name()) {
             Some(support::profile::only(profiles::Profile {
-                name: Some(target.name().clone()),
+                addr: Some(target.clone().into()),
                 ..profiles::Profile::default()
             }))
         } else {

linkerd/app/inbound/src/allow_discovery.rs

@@ -1,18 +1,18 @@
 use crate::target::Target;
 use linkerd_app_core::{
-    discovery_rejected, profiles::LogicalAddr, svc::stack::Predicate, Error, NameMatch,
+    discovery_rejected, profiles::LookupAddr, svc::stack::Predicate, Error, NameMatch,
 };
 
 #[derive(Clone, Debug)]
 pub struct AllowProfile(pub NameMatch);
 
 impl Predicate<Target> for AllowProfile {
-    type Request = LogicalAddr;
+    type Request = LookupAddr;
 
-    fn check(&mut self, target: Target) -> Result<LogicalAddr, Error> {
+    fn check(&mut self, target: Target) -> Result<LookupAddr, Error> {
         let addr = target.dst.into_name_addr().ok_or_else(discovery_rejected)?;
         if self.0.matches(addr.name()) {
-            Ok(LogicalAddr(addr.into()))
+            Ok(LookupAddr(addr.into()))
         } else {
             Err(discovery_rejected().into())
         }

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

@@ -115,7 +115,7 @@ where
             > + Clone,
     >
     where
-        P: profiles::GetProfile<profiles::LogicalAddr> + Clone + Send + Sync + 'static,
+        P: profiles::GetProfile<profiles::LookupAddr> + Clone + Send + Sync + 'static,
         P::Future: Send,
         P::Error: Send,
     {

linkerd/app/inbound/src/lib.rs

@@ -135,7 +135,7 @@ impl Inbound<()> {
             + 'static,
         GSvc::Error: Into<Error>,
         GSvc::Future: Send,
-        P: profiles::GetProfile<profiles::LogicalAddr> + Clone + Send + Sync + 'static,
+        P: profiles::GetProfile<profiles::LookupAddr> + Clone + Send + Sync + 'static,
         P::Error: Send,
         P::Future: Send,
     {
@@ -225,7 +225,7 @@ where
         GSvc: svc::Service<direct::GatewayIo<I>, Response = ()> + Send + 'static,
         GSvc::Error: Into<Error>,
         GSvc::Future: Send,
-        P: profiles::GetProfile<profiles::LogicalAddr> + Clone + Send + Sync + 'static,
+        P: profiles::GetProfile<profiles::LookupAddr> + Clone + Send + Sync + 'static,
         P::Error: Send,
         P::Future: Send,
     {

linkerd/app/inbound/src/target.rs

@@ -207,9 +207,9 @@ impl From<Logical> for Target {
     }
 }
 
-impl Param<profiles::LogicalAddr> for Target {
-    fn param(&self) -> profiles::LogicalAddr {
-        profiles::LogicalAddr(self.dst.clone())
+impl Param<profiles::LookupAddr> for Target {
+    fn param(&self) -> profiles::LookupAddr {
+        profiles::LookupAddr(self.dst.clone())
     }
 }
 

linkerd/app/outbound/src/discover.rs

@@ -25,7 +25,7 @@ impl<N> Outbound<N> {
         N: svc::NewService<tcp::Logical, Service = NSvc> + Clone + Send + 'static,
         NSvc: svc::Service<SensorIo<I>, Response = (), Error = Error> + Send + 'static,
         NSvc::Future: Send,
-        P: profiles::GetProfile<profiles::LogicalAddr> + Clone + Send + 'static,
+        P: profiles::GetProfile<profiles::LookupAddr> + Clone + Send + 'static,
         P::Future: Send,
         P::Error: Send,
     {
@@ -73,11 +73,11 @@ struct AllowProfile(pub IpMatch);
 // === impl AllowProfile ===
 
 impl svc::stack::Predicate<tcp::Accept> for AllowProfile {
-    type Request = profiles::LogicalAddr;
+    type Request = profiles::LookupAddr;
 
-    fn check(&mut self, a: tcp::Accept) -> Result<profiles::LogicalAddr, Error> {
+    fn check(&mut self, a: tcp::Accept) -> Result<profiles::LookupAddr, Error> {
         if self.0.matches(a.orig_dst.0.ip()) {
-            Ok(profiles::LogicalAddr(a.orig_dst.0.into()))
+            Ok(profiles::LookupAddr(a.orig_dst.0.into()))
         } else {
             Err(discovery_rejected().into())
         }

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

@@ -52,6 +52,7 @@ impl<C> Outbound<C> {
         // HTTP/1.x fallback is supported as needed.
         let stack = connect
             .push(http::client::layer(h1_settings, h2_settings))
+            .check_service::<T>()
             // Re-establishes a connection when the client fails.
             .push(reconnect::layer({
                 let backoff = backoff;

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

@@ -178,7 +178,7 @@ impl<E> Outbound<E> {
                     let should_resolve = match logical.profile.as_ref() {
                         Some(p) => {
                             let p = p.borrow();
-                            p.endpoint.is_none() && (p.name.is_some() || !p.targets.is_empty())
+                            p.endpoint.is_none() && (p.addr.is_some() || !p.targets.is_empty())
                         }
                         None => false,
                     };