inbound: connections wait for ServerPolicy discovery (#2186)

Currently, when the inbound proxy gets a connection to a port it has not
previously configured, it starts doing ServerPolicy discovery for that
port but then uses the default policy until the configuration is synced
from the control plane. This behavior is fairly complex to reason about
and surprising for users. Therefore, we would like to remove this
defaulting behavior from the inbound proxy.

This branch changes how the inbound proxy performs policy discovery.
Rather than spawning the lookup in a background task and using a default
policy until a policy is discovered, the inbound proxy will now wait
until the policy for a port is discovered before continuing to process a
connection on that port. In terms of the inbound stack, this means that
policy discovery is now a `MakeService` rather than a `NewService`. The
cache is still used when a policy discovery watch has already been
started.

This branch does not *completely* remove all default policy
configuration from the proxy. The default policy environment variables
are still used when the proxy is configured without a control plane
policy controller address. However, this mode will be removed in a
subsequent PR.

Author: hawkw

Cargo.lock

@@ -891,6 +891,7 @@ dependencies = [
  "linkerd2-proxy-api",
  "once_cell",
  "parking_lot",
+ "pin-project",
  "thiserror",
  "tokio",
  "tokio-test",

linkerd/app/admin/src/stack.rs

@@ -69,11 +69,15 @@ struct Rescue;
 // === impl Config ===
 
 impl Config {
+    /// Builds the admin endpoint server.
+    ///
+    /// This method is asynchronous, as it must discover a `ServerPolicy` for
+    /// the admin port.
     #[allow(clippy::too_many_arguments)]
-    pub fn build<B, R>(
+    pub async fn build<B, R>(
         self,
         bind: B,
-        policy: impl inbound::policy::GetPolicy,
+        policy: &impl inbound::policy::GetPolicy,
         identity: identity::Server,
         report: R,
         metrics: inbound::Metrics,
@@ -89,7 +93,7 @@ impl Config {
         let (listen_addr, listen) = bind.bind(&self.server)?;
 
         // Get the policy for the admin server.
-        let policy = policy.get_policy(OrigDstAddr(listen_addr.into()));
+        let policy = policy.get_policy(OrigDstAddr(listen_addr.into())).await?;
 
         let (ready, latch) = crate::server::Readiness::new();
         let admin = crate::server::Admin::new(report, ready, shutdown, trace);

linkerd/app/inbound/Cargo.toml

@@ -31,6 +31,7 @@ linkerd-tonic-watch = { path = "../../tonic-watch" }
 linkerd2-proxy-api = { version = "0.8", features = ["inbound"] }
 once_cell = "1"
 parking_lot = "0.12"
+pin-project = "1"
 thiserror = "1"
 tokio = { version = "1", features = ["sync"] }
 tonic = { version = "0.8", default-features = false }

linkerd/app/inbound/src/accept.rs

@@ -1,5 +1,5 @@
 use crate::{
-    policy::{AllowPolicy, GetPolicy},
+    policy::{self, AllowPolicy, GetPolicy},
     Inbound,
 };
 use linkerd_app_core::{
@@ -26,7 +26,7 @@ impl<N> Inbound<N> {
     pub(crate) fn push_accept<T, I, NSvc, D, DSvc>(
         self,
         proxy_port: u16,
-        policies: impl GetPolicy + Clone + Send + Sync + 'static,
+        policies: impl GetPolicy,
         direct: D,
     ) -> Inbound<svc::ArcNewTcp<T, I>>
     where
@@ -46,6 +46,19 @@ impl<N> Inbound<N> {
     {
         self.map_stack(|cfg, rt, accept| {
             accept
+                .push_on_service(svc::MapErr::layer_boxed())
+                .push_map_target(|(policy, t): (AllowPolicy, T)| {
+                    tracing::debug!(policy = ?&*policy.borrow(), "Accepted");
+                    Accept {
+                        client_addr: t.param(),
+                        orig_dst_addr: t.param(),
+                        policy,
+                    }
+                })
+                .lift_new_with_target()
+                .push(policy::Discover::layer(policies))
+                .into_new_service()
+                .check_new_service::<T, I>()
                 .push_switch(
                     // Switch to the `direct` stack when a connection's original destination is the
                     // proxy's inbound port. Otherwise, check that connections are allowed on the
@@ -56,13 +69,7 @@ impl<N> Inbound<N> {
                             return Ok(svc::Either::B(t));
                         }
 
-                        let policy = policies.get_policy(addr);
-                        tracing::debug!(policy = ?&*policy.borrow(), "Accepted");
-                        Ok(svc::Either::A(Accept {
-                            client_addr: t.param(),
-                            orig_dst_addr: addr,
-                            policy,
-                        }))
+                        Ok(svc::Either::A(t))
                     },
                     direct,
                 )

linkerd/app/inbound/src/direct.rs

@@ -91,7 +91,7 @@ impl<N> Inbound<N> {
     ///    gateways may need to accept HTTP requests from older proxy versions
     pub(crate) fn push_direct<T, I, NSvc, G, GSvc, H, HSvc>(
         self,
-        policies: impl policy::GetPolicy + Clone + Send + Sync + 'static,
+        policies: impl policy::GetPolicy,
         gateway: G,
         http: H,
     ) -> Inbound<svc::ArcNewTcp<T, I>>
@@ -146,58 +146,50 @@ impl<N> Inbound<N> {
                 // the header.
                 .push_switch(Ok::<Local, Infallible>, http)
                 .push_switch(
-                    {
-                        let policies = policies.clone();
-                        move |(h, client): (TransportHeader, ClientInfo)| -> Result<_, Infallible> {
-                            match h {
-                                TransportHeader {
-                                    port,
-                                    name: None,
-                                    protocol,
-                                } => Ok(svc::Either::A({
-                                    // When the transport header targets an alternate port (but does
-                                    // not identify an alternate target name), we check the new
-                                    // target's policy (rather than the inbound proxy's address).
-                                    let addr = (client.local_addr.ip(), port).into();
-                                    let policy = policies.get_policy(OrigDstAddr(addr));
-                                    match protocol {
-                                        None => svc::Either::A(LocalTcp {
-                                            server_addr: Remote(ServerAddr(addr)),
-                                            client_addr: client.client_addr,
-                                            client_id: client.client_id,
+                    |(policy, (h, client)): (AllowPolicy, (TransportHeader, ClientInfo))| -> Result<_, Infallible> {
+                        match h {
+                            TransportHeader {
+                                port,
+                                name: None,
+                                protocol,
+                             } => {
+                                // The transport header targets an alternate port (but does
+                                // not identify an alternate target name).
+                                let addr = (client.local_addr.ip(), port).into();
+                                Ok(svc::Either::A(match protocol {
+                                    None => svc::Either::A(LocalTcp {
+                                        server_addr: Remote(ServerAddr(addr)),
+                                        client_addr: client.client_addr,
+                                        client_id: client.client_id,
+                                        policy,
+                                    }),
+                                    Some(protocol) => {
+                                        // When TransportHeader includes the protocol, but does not
+                                        // include an alternate name we go through the Inbound HTTP
+                                        // stack.
+                                        svc::Either::B(LocalHttp {
+                                            addr: Remote(ServerAddr(addr)),
                                             policy,
-                                        }),
-                                        Some(protocol) => {
-                                            // When TransportHeader includes the protocol, but does not
-                                            // include an alternate name we go through the Inbound HTTP
-                                            // stack.
-                                            svc::Either::B(LocalHttp {
-                                                addr: Remote(ServerAddr(addr)),
-                                                policy,
-                                                protocol,
-                                                client,
-                                            })
-                                        }
+                                            protocol,
+                                            client,
+                                        })
                                     }
-                                })),
-
-                                TransportHeader {
-                                    port,
-                                    name: Some(name),
+                                }))
+                            },
+                            TransportHeader {
+                                port,
+                                name: Some(name),
+                                protocol,
+                            } => Ok(svc::Either::B(
+                                // When the transport header provides an alternate target, the
+                                // connection is a gateway connection.
+                                GatewayTransportHeader {
+                                    target: NameAddr::from((name, port)),
                                     protocol,
-                                } => Ok(svc::Either::B({
-                                    // When the transport header provides an alternate target, the
-                                    // connection is a gateway connection. We check the _gateway
-                                    // address's_ policy (rather than the target address).
-                                    let policy = policies.get_policy(client.local_addr);
-                                    GatewayTransportHeader {
-                                        target: NameAddr::from((name, port)),
-                                        protocol,
-                                        client,
-                                        policy,
-                                    }
-                                })),
-                            }
+                                    client,
+                                    policy,
+                                }
+                            ))
                         }
                     },
                     // HTTP detection is not necessary in this case, since the transport
@@ -211,6 +203,22 @@ impl<N> Inbound<N> {
                         )
                         .into_inner(),
                 )
+                .lift_new_with_target()
+                .push(policy::Discover::layer_via(
+                    policies,
+                    |(header, client): &(TransportHeader, ClientInfo)| {
+                        if header.name.is_some() {
+                            // When the transport header provides an alternate target, the
+                            // connection is a gateway connection. . We use the `OrigDstAddr`--the
+                            // inbound proxy server's address--to lookup policies.
+                            return client.local_addr;
+                        }
+
+                        // Otherwise, use the port override from the transport header.
+                        OrigDstAddr((client.local_addr.ip(), header.port).into())
+                    },
+                ))
+                .into_new_service()
                 .check_new_service::<(TransportHeader, ClientInfo), _>()
                 // Use ALPN to determine whether a transport header should be read.
                 .push(NewTransportHeaderServer::layer(detect_timeout))

linkerd/app/inbound/src/policy.rs

@@ -1,13 +1,15 @@
 mod api;
 mod config;
 pub mod defaults;
+mod discover;
 mod http;
 mod store;
 mod tcp;
 
 pub(crate) use self::store::Store;
 pub use self::{
     config::Config,
+    discover::Discover,
     http::{
         HttpInvalidPolicy, HttpRouteInvalidRedirect, HttpRouteNotFound, HttpRouteRedirect,
         HttpRouteUnauthorized, NewHttpPolicy,
@@ -20,6 +22,7 @@ use linkerd_app_core::{
     metrics::{RouteAuthzLabels, ServerAuthzLabels},
     tls,
     transport::{ClientAddr, OrigDstAddr, Remote},
+    Error,
 };
 use linkerd_idle_cache::Cached;
 pub use linkerd_proxy_server_policy::{
@@ -28,7 +31,7 @@ pub use linkerd_proxy_server_policy::{
     http::{filter::Redirection, Route as HttpRoute},
     route, Authentication, Authorization, Meta, Protocol, RoutePolicy, ServerPolicy,
 };
-use std::sync::Arc;
+use std::{future::Future, sync::Arc};
 use thiserror::Error;
 use tokio::sync::watch;
 
@@ -38,9 +41,11 @@ pub struct ServerUnauthorized {
     server: Arc<Meta>,
 }
 
-pub trait GetPolicy {
-    // Returns the traffic policy configured for the destination address.
-    fn get_policy(&self, dst: OrigDstAddr) -> AllowPolicy;
+/// Returns the traffic policy configured for the destination address.
+pub trait GetPolicy: Clone + Send + Sync + 'static {
+    type Future: Future<Output = Result<AllowPolicy, Error>> + Unpin + Send;
+
+    fn get_policy(&self, target: OrigDstAddr) -> Self::Future;
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -75,6 +80,24 @@ pub enum Routes {
     Grpc(Arc<[GrpcRoute]>),
 }
 
+// === impl GetPolicy ===
+
+impl<S> GetPolicy for S
+where
+    S: tower::Service<OrigDstAddr, Response = AllowPolicy, Error = Error>,
+    S: Clone + Send + Sync + Unpin + 'static,
+    S::Future: Send + Unpin,
+{
+    type Future = tower::util::Oneshot<S, OrigDstAddr>;
+
+    #[inline]
+    fn get_policy(&self, target: OrigDstAddr) -> Self::Future {
+        use tower::util::ServiceExt;
+
+        self.clone().oneshot(target)
+    }
+}
+
 // === impl DefaultPolicy ===
 
 impl From<ServerPolicy> for DefaultPolicy {

linkerd/app/inbound/src/policy/config.rs

@@ -36,7 +36,7 @@ impl Config {
         dns: dns::Resolver,
         metrics: metrics::ControlHttp,
         identity: identity::NewClient,
-    ) -> impl GetPolicy + Clone + Send + Sync + 'static {
+    ) -> impl GetPolicy {
         match self {
             Self::Fixed {
                 default,

linkerd/app/inbound/src/policy/discover.rs

@@ -0,0 +1,88 @@
+use super::{AllowPolicy, GetPolicy};
+use futures::ready;
+use linkerd_app_core::{svc, transport::OrigDstAddr, Error};
+use std::{
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+#[derive(Debug, Clone)]
+pub struct Discover<X, G, N> {
+    extract_addr: X,
+    get_policy: G,
+    new_svc: N,
+}
+
+#[pin_project::pin_project]
+pub struct DiscoverFuture<F, N> {
+    #[pin]
+    inner: F,
+    new_svc: N,
+}
+
+impl<G, N> Discover<(), G, N>
+where
+    G: GetPolicy + Clone,
+{
+    pub fn layer(get_policy: G) -> impl svc::layer::Layer<N, Service = Self> + Clone {
+        Self::layer_via(get_policy, ())
+    }
+}
+
+impl<X, G, N> Discover<X, G, N>
+where
+    G: GetPolicy + Clone,
+    X: Clone,
+{
+    pub fn layer_via(
+        get_policy: G,
+        extract_addr: X,
+    ) -> impl svc::layer::Layer<N, Service = Self> + Clone {
+        svc::layer::mk(move |new_svc| Self {
+            extract_addr: extract_addr.clone(),
+            get_policy: get_policy.clone(),
+            new_svc,
+        })
+    }
+}
+
+impl<X, G, N, NSvc, T> svc::Service<T> for Discover<X, G, N>
+where
+    G: GetPolicy,
+    X: svc::ExtractParam<OrigDstAddr, T>,
+    N: svc::NewService<T, Service = NSvc> + Clone,
+    NSvc: svc::NewService<AllowPolicy>,
+{
+    type Error = Error;
+    type Response = NSvc::Service;
+    type Future = DiscoverFuture<G::Future, NSvc>;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, target: T) -> Self::Future {
+        let dst = self.extract_addr.extract_param(&target);
+        DiscoverFuture {
+            inner: self.get_policy.get_policy(dst),
+            new_svc: self.new_svc.new_service(target),
+        }
+    }
+}
+
+// === impl DiscoverFuture ===
+
+impl<F, N, E> Future for DiscoverFuture<F, N>
+where
+    F: Future<Output = Result<AllowPolicy, E>>,
+    N: svc::NewService<AllowPolicy>,
+{
+    type Output = Result<N::Service, E>;
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+        let policy = ready!(this.inner.poll(cx))?;
+        let svc = this.new_svc.new_service(policy);
+        Poll::Ready(Ok(svc))
+    }
+}