gateway: synthesize ClientPolicies when the controller returns NotFound (#2333)

Both outbound and gateway proxies now resolve client policies from the
OutboundPolicies API. When the outbound proxy attempts to discover a
policy and the policy controller returns NotFound, it synthesizes a
default policy from the discovered ServiceProfile. However, when the
gateway proxy receives a NotFound, it will currently fail the
connection, based on the assumption that only valid cluster DNS names
are gatewayed (and not arbitrary IPs that might be forwards).
Unfortunately, this is not quite true. Gateway proxies may attempt to
discover cluster DNS names that are Pod DNS names, rather than Service
DNS names, and the policy controller will return NotFound for those
names.

This branch therefore changes the gateway proxy to also synthesize
default ClientPolicies based on the ServiceProfile when receiving a
NotFound status. Some of the code for synthesizing a client policy
from a ServiceProfile that's currently used in the outbound proxy was
factored out so that it could be reused here.

Author: hawkw

Cargo.lock

@@ -949,7 +949,9 @@ dependencies = [
  "linkerd-app-inbound",
  "linkerd-app-outbound",
  "linkerd-app-test",
+ "linkerd-proxy-client-policy",
  "linkerd-proxy-server-policy",
+ "once_cell",
  "thiserror",
  "tokio",
  "tokio-test",

linkerd/app/gateway/Cargo.toml

@@ -12,6 +12,8 @@ futures = { version = "0.3", default-features = false }
 linkerd-app-core = { path = "../core" }
 linkerd-app-inbound = { path = "../inbound" }
 linkerd-app-outbound = { path = "../outbound" }
+linkerd-proxy-client-policy = { path = "../../proxy/client-policy" }
+once_cell = "1"
 thiserror = "1"
 tokio = { version = "1", features = ["sync"] }
 tonic = { version = "0.8", default-features = false }

linkerd/app/gateway/src/discover.rs

@@ -1,8 +1,11 @@
 use crate::Gateway;
-use futures::TryFutureExt;
+use futures::FutureExt;
 use linkerd_app_core::{errors, profiles, svc, Error};
 use linkerd_app_inbound::{GatewayAddr, GatewayDomainInvalid};
 use linkerd_app_outbound::{self as outbound};
+use linkerd_proxy_client_policy::{self as policy, ClientPolicy};
+use once_cell::sync::Lazy;
+use std::sync::Arc;
 use tokio::sync::watch;
 use tracing::Instrument;
 
@@ -30,6 +33,14 @@ impl Gateway {
         use futures::future;
 
         let allowlist = self.config.allow_discovery.clone();
+        let detect_timeout = self.outbound.config().proxy.detect_protocol_timeout;
+        let queue = {
+            let queue = self.outbound.config().tcp_connection_queue;
+            policy::Queue {
+                capacity: queue.capacity,
+                failfast_timeout: queue.failfast_timeout,
+            }
+        };
         svc::mk(move |GatewayAddr(addr)| {
             tracing::debug!(%addr, "Discover");
 
@@ -45,19 +56,55 @@ impl Gateway {
 
             let policy = policies
                 .get_policy(addr.into())
-                .map_err(|e| {
-                    // If the policy controller returned `NotFound`, indicating
-                    // that it doesn't have a policy for this addr, then we
-                    // can't gateway this address.
-                    if is_not_found(&e) {
-                        GatewayDomainInvalid.into()
-                    } else {
-                        e
-                    }
-                })
                 .instrument(tracing::debug_span!("policy"));
 
-            future::Either::Right(future::try_join(profile, policy))
+            let discovery = future::join(profile, policy).map(move |(profile, policy)| {
+                tracing::debug!("Discovered");
+
+                let profile = profile?.ok_or(GatewayDomainInvalid)?;
+
+                // If there was a policy resolution, return it with the profile so
+                // the stack can determine how to switch on them.
+                match policy {
+                    Ok(policy) => return Ok((Some(profile), policy)),
+                    // The policy controller currently rejects discovery for DNS
+                    // names that are not Services, so we will get a `NotFound`
+                    // error if we looked up a pod DNS name. In this case, we
+                    // will synthesize a default policy.
+                    Err(error) if is_not_found(&error) => tracing::debug!("Policy not found"),
+                    Err(error) => return Err(error),
+                }
+
+                let policy = outbound::spawn_synthesized_profile_policy(
+                    profile.clone().into(),
+                    move |profile| {
+                        static META: Lazy<Arc<policy::Meta>> = Lazy::new(|| {
+                            Arc::new(policy::Meta::Default {
+                                name: "gateway".into(),
+                            })
+                        });
+
+                        match profile.endpoint.clone() {
+                            Some((addr, meta)) => outbound::synthesize_forward_policy(
+                                &META,
+                                detect_timeout,
+                                queue,
+                                addr,
+                                meta,
+                            ),
+                            None => {
+                                tracing::debug!(
+                                    "Gateway ServiceProfile does not contain an endpoint"
+                                );
+                                ClientPolicy::invalid(detect_timeout)
+                            }
+                        }
+                    },
+                );
+                Ok((Some(profile), policy))
+            });
+
+            future::Either::Right(discovery)
         })
     }
 }

linkerd/app/outbound/src/discover.rs

@@ -1,6 +1,8 @@
-use crate::{policy, Outbound};
+use crate::{
+    policy::{self, ClientPolicy},
+    Outbound,
+};
 use linkerd_app_core::{errors, profiles, svc, transport::OrigDstAddr, Error};
-use linkerd_proxy_client_policy::ClientPolicy;
 use once_cell::sync::Lazy;
 use std::{
     fmt::Debug,
@@ -114,10 +116,21 @@ impl<N> Outbound<N> {
                 // enpdoint policy.
                 if let Some(profile) = profile {
                     let policy = spawn_synthesized_profile_policy(
-                        orig_dst,
                         profile.clone().into(),
-                        queue,
-                        detect_timeout,
+                        move |profile: &profiles::Profile| {
+                            static META: Lazy<Arc<policy::Meta>> = Lazy::new(|| {
+                                Arc::new(policy::Meta::Default {
+                                    name: "endpoint".into(),
+                                })
+                            });
+                            let (addr, meta) = profile
+                                .endpoint
+                                .clone()
+                                .unwrap_or_else(|| (orig_dst, Default::default()));
+                            // TODO(ver) We should be able to figure out resource coordinates for
+                            // the endpoint?
+                            synthesize_forward_policy(&META, detect_timeout, queue, addr, meta)
+                        },
                     );
                     return Ok((Some(profile), policy));
                 }
@@ -137,29 +150,11 @@ fn is_not_found(e: &Error) -> bool {
         .unwrap_or(false)
 }
 
-fn spawn_synthesized_profile_policy(
-    orig_dst: SocketAddr,
+pub fn spawn_synthesized_profile_policy(
     mut profile: watch::Receiver<profiles::Profile>,
-    queue: policy::Queue,
-    detect_timeout: Duration,
+    synthesize: impl Fn(&profiles::Profile) -> policy::ClientPolicy + Send + 'static,
 ) -> watch::Receiver<policy::ClientPolicy> {
-    static META: Lazy<Arc<policy::Meta>> = Lazy::new(|| {
-        Arc::new(policy::Meta::Default {
-            name: "endpoint".into(),
-        })
-    });
-
-    let mk = move |profile: &profiles::Profile| {
-        let (addr, meta) = profile
-            .endpoint
-            .clone()
-            .unwrap_or_else(|| (orig_dst, Default::default()));
-        // TODO(ver) We should be able to figure out resource coordinates for
-        // the endpoint?
-        synthesize_forward_policy(&META, detect_timeout, queue, addr, meta)
-    };
-
-    let policy = mk(&*profile.borrow_and_update());
+    let policy = synthesize(&*profile.borrow_and_update());
     tracing::debug!(?policy, profile = ?*profile.borrow(), "Synthesizing policy from profile");
     let (tx, rx) = watch::channel(policy);
     tokio::spawn(
@@ -178,7 +173,7 @@ fn spawn_synthesized_profile_policy(
                         }
                     }
                 };
-                let policy = mk(&*profile.borrow());
+                let policy = synthesize(&*profile.borrow());
                 tracing::debug!(?policy, "Profile updated; synthesizing policy");
                 if tx.send(policy).is_err() {
                     tracing::debug!("Policy watch closed, terminating");
@@ -191,28 +186,7 @@ fn spawn_synthesized_profile_policy(
     rx
 }
 
-fn spawn_synthesized_origdst_policy(
-    orig_dst: SocketAddr,
-    queue: policy::Queue,
-    detect_timeout: Duration,
-) -> watch::Receiver<policy::ClientPolicy> {
-    static META: Lazy<Arc<policy::Meta>> = Lazy::new(|| {
-        Arc::new(policy::Meta::Default {
-            name: "fallback".into(),
-        })
-    });
-
-    let policy =
-        synthesize_forward_policy(&META, detect_timeout, queue, orig_dst, Default::default());
-    tracing::debug!(?policy, "Synthesizing policy");
-    let (tx, rx) = watch::channel(policy);
-    tokio::spawn(async move {
-        tx.closed().await;
-    });
-    rx
-}
-
-fn synthesize_forward_policy(
+pub fn synthesize_forward_policy(
     meta: &Arc<policy::Meta>,
     timeout: Duration,
     queue: policy::Queue,
@@ -273,6 +247,27 @@ fn synthesize_forward_policy(
     }
 }
 
+fn spawn_synthesized_origdst_policy(
+    orig_dst: SocketAddr,
+    queue: policy::Queue,
+    detect_timeout: Duration,
+) -> watch::Receiver<policy::ClientPolicy> {
+    static META: Lazy<Arc<policy::Meta>> = Lazy::new(|| {
+        Arc::new(policy::Meta::Default {
+            name: "fallback".into(),
+        })
+    });
+
+    let policy =
+        synthesize_forward_policy(&META, detect_timeout, queue, orig_dst, Default::default());
+    tracing::debug!(?policy, "Synthesizing policy");
+    let (tx, rx) = watch::channel(policy);
+    tokio::spawn(async move {
+        tx.closed().await;
+    });
+    rx
+}
+
 // === impl Discovery ===
 
 impl<T> From<((Option<profiles::Receiver>, policy::Receiver), T)> for Discovery<T> {

linkerd/app/outbound/src/lib.rs

@@ -44,7 +44,10 @@ pub mod tcp;
 #[cfg(any(test, feature = "test-util"))]
 pub mod test_util;
 
-pub use self::{discover::Discovery, metrics::Metrics};
+pub use self::{
+    discover::{spawn_synthesized_profile_policy, synthesize_forward_policy, Discovery},
+    metrics::Metrics,
+};
 
 #[derive(Clone, Debug)]
 pub struct Config {