inbound: connections wait for ServerPolicy discovery

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,18 @@ 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,
+                    }
+                })
+                .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 +68,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

@@ -80,6 +80,13 @@ struct TlsParams {
     identity: identity::Server,
 }
 
+/// A policy discovery target with a transport header and client info.
+#[derive(Debug, Clone)]
+struct Discover {
+    header: TransportHeader,
+    client: ClientInfo,
+}
+
 impl<N> Inbound<N> {
     /// Builds a stack that handles connections that target the proxy's inbound port
     /// (i.e. without an SO_ORIGINAL_DST setting). This port behaves differently from
@@ -91,7 +98,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,59 +153,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 {
+                    |(policy, target): (AllowPolicy, Discover)| -> Result<_, Infallible> {
+                        let Discover {
+                            header:
                                 TransportHeader {
+                                    name,
                                     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,
-                                        }),
-                                        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,
-                                            })
-                                        }
-                                    }
-                                })),
+                                },
+                            client,
+                        } = target;
+
+                        if let Some(name) = name {
+                            // When the transport header provides an alternate target, the
+                            // connection is a gateway connection.
+                            return Ok(svc::Either::B(GatewayTransportHeader {
+                                target: NameAddr::from((name, port)),
+                                protocol,
+                                client,
+                                policy,
+                            }));
+                        }
 
-                                TransportHeader {
-                                    port,
-                                    name: Some(name),
+                        // 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,
                                     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,
+                                })
                             }
-                        }
+                        }))
                     },
                     // HTTP detection is not necessary in this case, since the transport
                     // header indicates the connection's HTTP version.
@@ -211,6 +209,9 @@ impl<N> Inbound<N> {
                         )
                         .into_inner(),
                 )
+                .push(policy::Discover::layer(policies))
+                .into_new_service()
+                .push_map_target(|(header, client)| Discover { header, client })
                 .check_new_service::<(TransportHeader, ClientInfo), _>()
                 // Use ALPN to determine whether a transport header should be read.
                 .push(NewTransportHeaderServer::layer(detect_timeout))
@@ -484,3 +485,21 @@ impl<T> InsertParam<tls::ConditionalServerTls, T> for TlsParams {
         (tls, target)
     }
 }
+
+// === impl Discover ===
+
+impl svc::Param<OrigDstAddr> for Discover {
+    fn param(&self) -> OrigDstAddr {
+        if self.header.name.is_none() {
+            // 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).
+            return OrigDstAddr((self.client.local_addr.ip(), self.header.port).into());
+        }
+
+        // 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).
+        self.client.local_addr
+    }
+}

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)]

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,73 @@
+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<G, N> {
+    get_policy: G,
+    new_svc: N,
+}
+
+#[pin_project::pin_project]
+pub struct DiscoverFuture<T, F, N> {
+    target: Option<T>,
+    #[pin]
+    inner: F,
+    new_svc: N,
+}
+
+impl<G: GetPolicy + Clone, N> Discover<G, N> {
+    pub fn layer(get_policy: G) -> impl svc::layer::Layer<N, Service = Self> + Clone {
+        svc::layer::mk(move |new_svc| Self {
+            get_policy: get_policy.clone(),
+            new_svc,
+        })
+    }
+}
+
+impl<G: GetPolicy, N, T> svc::Service<T> for Discover<G, N>
+where
+    G: GetPolicy,
+    N: svc::NewService<(AllowPolicy, T)> + Clone,
+    T: svc::Param<OrigDstAddr>,
+{
+    type Error = Error;
+    type Response = N::Service;
+    type Future = DiscoverFuture<T, G::Future, N>;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, target: T) -> Self::Future {
+        let dst = target.param();
+        DiscoverFuture {
+            target: Some(target),
+            inner: self.get_policy.get_policy(dst),
+            new_svc: self.new_svc.clone(),
+        }
+    }
+}
+
+// === impl DiscoverFuture ===
+
+impl<T, F, N, E> Future for DiscoverFuture<T, F, N>
+where
+    F: Future<Output = Result<AllowPolicy, E>>,
+    N: svc::NewService<(AllowPolicy, T)>,
+{
+    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, this.target.take().expect("polled after ready")));
+        Poll::Ready(Ok(svc))
+    }
+}