inbound: Load policies lazily (#1249)

Rather than block inbound communication on policy discovery, this change
modifies policy initialization to use the default policy until policy
has been synced from the control plane. This will allow us to use
policies with the admin server.

Author: olix0r

linkerd/app/inbound/src/metrics/authz.rs

@@ -26,7 +26,7 @@ metrics! {
 }
 
 #[derive(Clone, Debug, Default)]
-pub(crate) struct HttpAuthzMetrics(Arc<HttpInner>);
+pub struct HttpAuthzMetrics(Arc<HttpInner>);
 
 #[derive(Clone, Debug, Default)]
 pub(crate) struct TcpAuthzMetrics(Arc<TcpInner>);

linkerd/app/inbound/src/policy/authorize/http.rs

@@ -36,9 +36,7 @@ pub struct AuthorizeHttp<T, N> {
 // === impl NewAuthorizeHttp ===
 
 impl<N> NewAuthorizeHttp<N> {
-    pub(crate) fn layer(
-        metrics: HttpAuthzMetrics,
-    ) -> impl svc::layer::Layer<N, Service = Self> + Clone {
+    pub fn layer(metrics: HttpAuthzMetrics) -> impl svc::layer::Layer<N, Service = Self> + Clone {
         svc::layer::mk(move |inner| Self {
             metrics: metrics.clone(),
             inner,

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

@@ -1,7 +1,5 @@
 use super::{discover::Discover, DefaultPolicy, ServerPolicy, Store};
-use linkerd_app_core::{
-    control, dns, metrics, proxy::identity::LocalCrtKey, svc::NewService, Result,
-};
+use linkerd_app_core::{control, dns, metrics, proxy::identity::LocalCrtKey, svc::NewService};
 use std::collections::{HashMap, HashSet};
 
 /// Configures inbound policies.
@@ -25,12 +23,12 @@ pub enum Config {
 // === impl Config ===
 
 impl Config {
-    pub(crate) async fn build(
+    pub(crate) fn build(
         self,
         dns: dns::Resolver,
         metrics: metrics::ControlHttp,
         identity: Option<LocalCrtKey>,
-    ) -> Result<Store> {
+    ) -> Store {
         match self {
             Self::Fixed { default, ports } => {
                 let (store, tx) = Store::fixed(default, ports);
@@ -39,7 +37,7 @@ impl Config {
                         tx.closed().await;
                     });
                 }
-                Ok(store)
+                store
             }
             Self::Discover {
                 control,
@@ -52,7 +50,7 @@ impl Config {
                     let c = control.build(dns, metrics, identity).new_service(());
                     Discover::new(workload, c).into_watch(backoff)
                 };
-                Store::spawn_discover(default, ports, watch).await
+                Store::spawn_discover(default, ports, watch)
             }
         }
     }

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

@@ -63,6 +63,19 @@ impl From<ServerPolicy> for DefaultPolicy {
     }
 }
 
+impl From<DefaultPolicy> for ServerPolicy {
+    fn from(d: DefaultPolicy) -> Self {
+        match d {
+            DefaultPolicy::Allow(p) => p,
+            DefaultPolicy::Deny => ServerPolicy {
+                protocol: Protocol::Opaque,
+                authorizations: vec![],
+                name: "default:deny".to_string(),
+            },
+        }
+    }
+}
+
 // === impl AllowPolicy ===
 
 impl AllowPolicy {

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

@@ -1,15 +1,13 @@
 use super::{discover, AllowPolicy, CheckPolicy, DefaultPolicy, DeniedUnknownPort};
-use futures::prelude::*;
 use linkerd_app_core::{proxy::http, transport::OrigDstAddr, Error, Result};
 pub use linkerd_server_policy::{Authentication, Authorization, Protocol, ServerPolicy, Suffix};
 use std::{
     collections::{HashMap, HashSet},
-    future::Future,
     hash::{BuildHasherDefault, Hasher},
     sync::Arc,
 };
 use tokio::sync::watch;
-use tracing::{info_span, Instrument};
+use tracing::info_span;
 
 #[derive(Clone, Debug)]
 pub struct Store {
@@ -74,49 +72,41 @@ impl Store {
     /// provided ports. The store maintains these watches so that each time a policy is checked, it
     /// may obtain the latest policy provided by the watch. An error is returned if any of the
     /// watches cannot be established.
-    //
-    // XXX(ver): rustc can't seem to figure out that this Future is `Send` unless we annotate it
-    // explicitly, hence the manual_async_fn.
-    #[allow(clippy::manual_async_fn)]
     pub(super) fn spawn_discover<S>(
         default: DefaultPolicy,
         ports: HashSet<u16>,
         discover: discover::Watch<S>,
-    ) -> impl Future<Output = Result<Self>> + Send
+    ) -> Self
     where
         S: tonic::client::GrpcService<tonic::body::BoxBody, Error = Error>,
         S: Clone + Send + Sync + 'static,
         S::Future: Send,
         S::ResponseBody: http::HttpBody<Error = Error> + Send + Sync + 'static,
     {
-        async move {
-            let rxs = ports.into_iter().map(|port| {
-                discover
-                    .clone()
-                    .spawn_watch(port)
-                    .instrument(info_span!("watch", %port))
-                    .map_ok(move |rsp| (port, rsp.into_inner()))
-            });
-
-            let ports = futures::future::join_all(rxs)
-                .await
-                .into_iter()
-                .collect::<Result<PortMap<_>, tonic::Status>>()?;
-
-            let default = match Self::mk_default(default) {
-                Some((tx, rx)) => {
-                    tokio::spawn(async move {
-                        tx.closed().await;
-                    });
-                    Some(rx)
-                }
-                None => None,
-            };
-
-            Ok(Self {
-                default,
-                ports: Arc::new(ports),
+        let rxs = ports
+            .into_iter()
+            .map(|port| {
+                let discover = discover.clone();
+                let default = default.clone();
+                let rx = info_span!("watch", %port)
+                    .in_scope(|| discover.spawn_with_init(port, default.into()));
+                (port, rx)
             })
+            .collect::<PortMap<_>>();
+
+        let default = match Self::mk_default(default) {
+            Some((tx, rx)) => {
+                tokio::spawn(async move {
+                    tx.closed().await;
+                });
+                Some(rx)
+            }
+            None => None,
+        };
+
+        Self {
+            default,
+            ports: Arc::new(rxs),
         }
     }
 }

linkerd/app/inbound/src/server.rs

@@ -16,7 +16,7 @@ struct TcpEndpoint {
 // === impl Inbound ===
 
 impl Inbound<()> {
-    pub async fn build_policies(
+    pub fn build_policies(
         &self,
         dns: dns::Resolver,
         control_metrics: metrics::ControlHttp,
@@ -25,8 +25,6 @@ impl Inbound<()> {
             .policy
             .clone()
             .build(dns, control_metrics, self.runtime.identity.clone())
-            .await
-            .expect("Failed to fetch port policy")
     }
 
     pub async fn serve<A, I, G, GSvc, P>(

linkerd/app/src/lib.rs

@@ -210,10 +210,8 @@ impl Config {
                         .instrument(info_span!("outbound")),
                 );
 
-                let inbound_policies = inbound
-                    .build_policies(dns, control_metrics)
-                    .instrument(info_span!("policy"))
-                    .await;
+                let inbound_policies =
+                    info_span!("policy").in_scope(|| inbound.build_policies(dns, control_metrics));
 
                 tokio::spawn(
                     inbound

linkerd/tonic-watch/src/lib.rs

@@ -48,7 +48,7 @@ where
         S: Service<T, Response = InnerRsp<U>, Error = tonic::Status>,
         S::Future: Send,
     {
-        // Get an update and stream or return None.
+        // Get an update and stream.
         let (init, rsp) = self.init(&target, None).await?;
 
         Ok(rsp.map(move |inner| {
@@ -60,6 +60,34 @@ where
         }))
     }
 
+    /// Returns a watch using the provided initial value.
+    ///
+    /// The watch is spawned on a background task that completes when the watch service errors in an
+    /// unrecoverable way or all receivers are dropped.
+    pub fn spawn_with_init<T, U>(mut self, target: T, init: U) -> watch::Receiver<U>
+    where
+        T: Clone + Send + Sync + 'static,
+        U: Send + Sync + 'static,
+        S: Service<T, Response = InnerRsp<U>, Error = tonic::Status>,
+        S::Future: Send,
+    {
+        let (tx, rx) = watch::channel(init);
+
+        // Spawn a background task to watch the inner service.
+        tokio::spawn(
+            async move {
+                let (up, rsp) = self.init(&target, None).await?;
+                if tx.send(up).is_ok() {
+                    self.publish_updates(target, tx, rsp.into_inner()).await;
+                }
+                Ok::<_, tonic::Status>(())
+            }
+            .in_current_span(),
+        );
+
+        rx
+    }
+
     /// Initiates a lookup stream and obtains the first profile from it.
     ///
     /// If the call fails, recovery and back-off are applied.