Handle profile endpoint in Gateway outbound stack (#1157)

In linkerd2/PR#6090, we noticed that when testing the StatefulSet
changes with a service that sends a request from the source cluster and
expects to be connected directly to a pod in the target cluster, the
outbound side of the Gateway in the target cluster throws a 'BadDomain'
error. When establishing a direct connection to a pod, the profile
returned from the look-up will contain an endpoint and it will not
contain a logical address.

The 'BadDomain' error stems from how the Gateway handles outbound
connections. If the profile does not include a logical address (i.e an
authorityOverride), then the connection is failed. This change adds
functionality to the Gateway to also consider endpoints received in the
profile on the outbound side. Support for endpoint handling on the
outbound side will unblock PR#6090.

To test, I have added a unit test that will check the endpoint stack is
considered when dealing with an endpoint in the profile. There are also
manual tests done in conjunction with the changes from PR#6090.

Signed-off-by: Matei David <[email protected]>

Author: mateiidavid

linkerd/app/gateway/src/gateway.rs

@@ -49,7 +49,10 @@ impl<O> NewGateway<O> {
 
 impl<O> svc::NewService<Target> for NewGateway<O>
 where
-    O: svc::NewService<outbound::http::Logical> + Send + Clone + 'static,
+    O: svc::NewService<svc::Either<outbound::http::Logical, outbound::http::Endpoint>>
+        + Send
+        + Clone
+        + 'static,
 {
     type Service = Gateway<O::Service>;
 
@@ -63,6 +66,23 @@ where
             None => return Gateway::BadDomain(http.target.name().clone()),
         };
 
+        // Create an outbound target using the endpoint from the profile.
+        if let Some((addr, metadata)) = profile.endpoint() {
+            debug!("Creating outbound endpoint");
+            let svc = self
+                .outbound
+                .new_service(svc::Either::B(outbound::http::Endpoint::from((
+                    http.version,
+                    outbound::tcp::Endpoint::from_metadata(
+                        addr,
+                        metadata,
+                        tls::NoClientTls::NotProvidedByServiceDiscovery,
+                        profile.is_opaque_protocol(),
+                    ),
+                ))));
+            return Gateway::new(svc, http.target, local_id);
+        }
+
         let logical_addr = match profile.logical_addr() {
             Some(addr) => addr,
             None => return Gateway::BadDomain(http.target.name().clone()),
@@ -72,11 +92,13 @@ where
         // including the original port. We don't know the IP of the target, so
         // we use an unroutable one.
         debug!("Creating outbound service");
-        let svc = self.outbound.new_service(outbound::http::Logical {
-            profile,
-            protocol: http.version,
-            logical_addr,
-        });
+        let svc = self
+            .outbound
+            .new_service(svc::Either::A(outbound::http::Logical {
+                profile,
+                protocol: http.version,
+                logical_addr,
+            }));
 
         Gateway::new(svc, http.target, local_id)
     }

linkerd/app/gateway/src/lib.rs

@@ -103,33 +103,51 @@ where
     // For each gatewayed connection that is *not* HTTP, use the target from the
     // transport header to lookup a service profile. If the profile includes a
     // resolvable service name, then continue with TCP endpoint resolution,
-    // balancing, and forwarding. An invalid original destination address is
-    // used so that service discovery is *required* to provide a valid endpoint.
+    // balancing, and forwarding. If the profile includes an endpoint instead
+    // of a logical address, then connect to endpoint directly and avoid
+    // balancing.
     //
     // TODO: We should use another target type that actually reflects
     // reality. But the outbound stack is currently pretty tightly
     // coupled to its target types.
-    let tcp = outbound
+    let logical = outbound
         .clone()
         .push_tcp_endpoint()
-        .push_tcp_logical(resolve.clone())
-        .into_stack()
-        .push_request_filter(
-            |(p, _): (Option<profiles::Receiver>, _)| -> Result<_, Error> {
-                let profile = p.ok_or_else(|| {
+        .push_tcp_logical(resolve.clone());
+    let endpoint = outbound
+        .clone()
+        .push_tcp_endpoint()
+        .push_tcp_forward()
+        .into_stack();
+    let tcp = endpoint
+        .push_switch(
+            move |(profile, _): (Option<profiles::Receiver>, _)| -> Result<_, Error> {
+                let profile = profile.ok_or_else(|| {
                     DiscoveryRejected::new("no profile discovered for gateway target")
                 })?;
+
+                if let Some((addr, metadata)) = profile.endpoint() {
+                    return Ok(svc::Either::A(outbound::tcp::Endpoint::from_metadata(
+                        addr,
+                        metadata,
+                        tls::NoClientTls::NotProvidedByServiceDiscovery,
+                        profile.is_opaque_protocol(),
+                    )));
+                }
+
                 let logical_addr = profile.logical_addr().ok_or_else(|| {
                     DiscoveryRejected::new(
-                        "profile for gateway target does not have a logical address",
+                        "profiles must have either an endpoint or a logical address",
                     )
                 })?;
-                Ok(outbound::tcp::Logical {
+
+                Ok(svc::Either::B(outbound::tcp::Logical {
                     profile,
                     protocol: (),
                     logical_addr,
-                })
+                }))
             },
+            logical.into_inner(),
         )
         .push(profiles::discover::layer(profiles.clone(), {
             let allow = allow_discovery.clone();
@@ -162,11 +180,12 @@ where
     // The client's ID is set as a request extension, as required by the
     // gateway. This permits gateway services (and profile resolutions) to be
     // cached per target, shared across clients.
-    let http = outbound
-        .push_tcp_endpoint()
-        .push_http_endpoint()
+    let endpoint = outbound.push_tcp_endpoint().push_http_endpoint();
+    let http = endpoint
+        .clone()
         .push_http_logical(resolve)
         .into_stack()
+        .push_switch(Ok::<_, Never>, endpoint.into_stack())
         .push(NewGateway::layer(local_id))
         .push(profiles::discover::layer(profiles, move |t: HttpTarget| {
             if allow_discovery.matches(t.target.name()) {

linkerd/app/gateway/src/tests.rs

@@ -11,7 +11,31 @@ use tower_test::mock;
 #[tokio::test]
 async fn gateway() {
     assert_eq!(
-        Test::default().run().await.unwrap().status(),
+        Test::default()
+            .with_default_profile()
+            .run()
+            .await
+            .unwrap()
+            .status(),
+        http::StatusCode::NO_CONTENT
+    );
+}
+
+#[tokio::test]
+async fn gateway_endpoint() {
+    let addr = std::net::SocketAddr::new([192, 0, 2, 10].into(), 777);
+    let profile = support::profile::only(profiles::Profile {
+        endpoint: Some((addr, Metadata::default())),
+        ..profiles::Profile::default()
+    });
+
+    assert_eq!(
+        Test::default()
+            .with_profile(profile)
+            .run()
+            .await
+            .unwrap()
+            .status(),
         http::StatusCode::NO_CONTENT
     );
 }
@@ -23,6 +47,7 @@ async fn bad_domain() {
         ..Default::default()
     };
     let status = test
+        .with_default_profile()
         .run()
         .await
         .unwrap_err()
@@ -39,6 +64,7 @@ async fn no_identity() {
         ..Default::default()
     };
     let status = test
+        .with_default_profile()
         .run()
         .await
         .unwrap_err()
@@ -57,6 +83,7 @@ async fn forward_loop() {
         ..Default::default()
     };
     let status = test
+        .with_default_profile()
         .run()
         .await
         .unwrap_err()
@@ -71,6 +98,7 @@ struct Test {
     target: NameAddr,
     client_id: Option<tls::ClientId>,
     orig_fwd: Option<&'static str>,
+    profile: Option<profiles::Receiver>,
 }
 
 impl Default for Test {
@@ -80,36 +108,31 @@ impl Default for Test {
             target: NameAddr::from_str("dst.test.example.com:4321").unwrap(),
             client_id: Some(tls::ClientId::from_str("client.id.test").unwrap()),
             orig_fwd: None,
+            profile: None,
         }
     }
 }
 
 impl Test {
     async fn run(self) -> Result<http::Response<http::BoxBody>, Error> {
         let Self {
-            suffix,
+            suffix: _,
             target,
             client_id,
             orig_fwd,
+            profile,
         } = self;
 
         let (outbound, mut handle) =
             mock::pair::<http::Request<http::BoxBody>, http::Response<http::BoxBody>>();
 
         let new = NewGateway::new(
-            move |_: outbound::http::Logical| outbound.clone(),
+            move |_: svc::Either<outbound::http::Logical, outbound::http::Endpoint>| {
+                outbound.clone()
+            },
             Some(tls::LocalId(id::Name::from_str("gateway.id.test").unwrap())),
         );
 
-        let allow = NameMatch::new(Some(dns::Suffix::from_str(suffix).unwrap()));
-        let profile = if allow.matches(target.name()) {
-            Some(support::profile::only(profiles::Profile {
-                addr: Some(target.clone().into()),
-                ..profiles::Profile::default()
-            }))
-        } else {
-            None
-        };
         let t = HttpTarget {
             target: target.clone(),
             version: http::Version::Http1,
@@ -146,4 +169,20 @@ impl Test {
         bg.await?;
         Ok(rsp)
     }
+
+    fn with_profile(mut self, profile: profiles::Receiver) -> Self {
+        let allow = NameMatch::new(Some(dns::Suffix::from_str(self.suffix).unwrap()));
+        if allow.matches(self.target.name()) {
+            self.profile = Some(profile);
+        }
+        self
+    }
+
+    fn with_default_profile(self) -> Self {
+        let target = self.target.clone();
+        self.with_profile(support::profile::only(profiles::Profile {
+            addr: Some(target.into()),
+            ..profiles::Profile::default()
+        }))
+    }
 }

linkerd/app/outbound/src/endpoint.rs

@@ -38,7 +38,7 @@ impl Endpoint<()> {
         }
     }
 
-    pub(crate) fn from_metadata(
+    pub fn from_metadata(
         addr: impl Into<SocketAddr>,
         metadata: Metadata,
         reason: tls::NoClientTls,

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

@@ -99,6 +99,7 @@ impl<E> Outbound<E> {
                 // If the traffic split is empty/unavailable, eagerly fail requests.
                 // When the split is in failfast, spawn the service in a background
                 // task so it becomes ready without new requests.
+                .check_new_service::<(ConcreteAddr, Logical), _>()
                 .push(profiles::split::layer())
                 .push_on_response(
                     svc::layers()