app: Block proxy initialization on identity readiness (#1202)

Currently, the proxy may start serving connections even before identity
is provisioned. This won't be feasible once we introduce policy
discovery--the proxy will have to establish its inbound policy before
serving inbound connections.

This change modifies proxy initialization to block proxy initialization
on identity. When identity isn't established, warnings are logged every
15s.

While this is a behavior change from the proxy's point of view, this
shouldn't be a user-facing change, as container initialization is
blocked on the proxy's readiness, which is dependent on identity
initialization.

Author: olix0r

Cargo.lock

@@ -1168,6 +1168,7 @@ dependencies = [
  "linkerd-tls",
  "linkerd2-proxy-api",
  "pin-project",
+ "thiserror",
  "tokio",
  "tonic",
  "tracing",

linkerd/app/src/identity.rs

@@ -8,8 +8,7 @@ use linkerd_app_core::{
     metrics::ControlHttp as Metrics,
     Error,
 };
-use std::future::Future;
-use std::pin::Pin;
+use std::{future::Future, pin::Pin};
 use tracing::Instrument;
 
 // The Disabled case is extraordinarily rare.
@@ -39,6 +38,8 @@ struct Recover(ExponentialBackoff);
 
 pub type Task = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
 
+// === impl Config ===
+
 impl Config {
     pub fn build(self, dns: dns::Resolver, metrics: Metrics) -> Result<Identity, Error> {
         match self {
@@ -55,7 +56,7 @@ impl Config {
                     Box::pin(
                         daemon
                             .run(svc)
-                            .instrument(tracing::debug_span!("identity_daemon", peer.addr = %addr)),
+                            .instrument(tracing::debug_span!("identity", server.addr = %addr)),
                     )
                 };
 
@@ -65,6 +66,8 @@ impl Config {
     }
 }
 
+// === impl Identity ===
+
 impl Identity {
     pub fn local(&self) -> Option<LocalCrtKey> {
         match self {
@@ -88,6 +91,8 @@ impl Identity {
     }
 }
 
+// === impl Recover ===
+
 impl<E: Into<Error>> linkerd_error::Recover<E> for Recover {
     type Backoff = ExponentialBackoffStream;
 

linkerd/app/src/lib.rs

@@ -203,12 +203,15 @@ impl Config {
 
         // Build a task that initializes and runs the proxy stacks.
         let start_proxy = {
+            let identity = identity.local();
             let inbound_addr = inbound_addr;
             let profiles = dst.profiles;
             let resolve = dst.resolve;
 
             Box::pin(async move {
-                // TODO(ver): Block on identity.
+                Self::await_identity(identity)
+                    .await
+                    .expect("failed to initialize identity");
 
                 tokio::spawn(
                     outbound
@@ -238,6 +241,30 @@ impl Config {
             tap,
         })
     }
+
+    /// Waits for the proxy's identity to be certified.
+    ///
+    /// If this does not complete in a timely fashion, warnings are logged every 15s
+    async fn await_identity(id: Option<identity::LocalCrtKey>) -> Result<(), Error> {
+        let id = match id {
+            Some(id) => id,
+            None => return Ok(()),
+        };
+
+        tokio::pin! {
+            let fut = id.await_crt();
+        }
+
+        const TIMEOUT: time::Duration = time::Duration::from_secs(15);
+        loop {
+            tokio::select! {
+                res = (&mut fut) => return res.map(|_| ()).map_err(Into::into),
+                _ = time::sleep(TIMEOUT) => {
+                    tracing::warn!("Waiting for identity to be initialized...");
+                }
+            }
+        }
+    }
 }
 
 impl App {

linkerd/proxy/identity/Cargo.toml

@@ -17,6 +17,7 @@ linkerd-identity = { path = "../../identity" }
 linkerd-metrics = { path = "../../metrics" }
 linkerd-stack = { path = "../../stack" }
 linkerd-tls = { path = "../../tls" }
+thiserror = "1"
 tokio = { version = "1", features = ["time", "sync"] }
 tonic = { version = "0.5", default-features = false }
 tracing = "0.1.26"

linkerd/proxy/identity/src/certify.rs

@@ -9,6 +9,7 @@ use pin_project::pin_project;
 use std::convert::TryFrom;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use thiserror::Error;
 use tokio::sync::watch;
 use tokio::time::{self, Sleep};
 use tonic::{self as grpc, body::BoxBody, client::GrpcService};
@@ -42,8 +43,9 @@ pub struct LocalCrtKey {
 #[derive(Debug)]
 pub struct AwaitCrt(Option<LocalCrtKey>);
 
-#[derive(Copy, Clone, Debug)]
-pub struct LostDaemon;
+#[derive(Copy, Clone, Debug, Error)]
+#[error("identity initialization failed")]
+pub struct LostDaemon(());
 
 pub type CrtKeySender = watch::Sender<Option<id::CrtKey>>;
 
@@ -187,7 +189,7 @@ impl LocalCrtKey {
         while self.crt_key.borrow().is_none() {
             // If the sender is dropped, the daemon task has ended.
             if self.crt_key.changed().await.is_err() {
-                return Err(LostDaemon);
+                return Err(LostDaemon(()));
             }
         }
         Ok(self)