app: Defer stack construction until the proxy starts (#1200)

olix0r · web-flow · commit 69dbf631de9d · 2021-08-11T11:48:07.000-07:00
We currently eagerly build the inbound and outbound stacks. In an
upcoming change, however, we'll want to ensure that we've obtained
inbound port policies before initializing the inbound stack.

This change modifies app initialization to move stack initialization
into the proxy's start task. In order to do this, we eagerly bind the
inbound/outbound listeners and provide the stream of incoming sockets to
the stacks at construction-time.

This doesn't actually change any behavior since stack construction was
moved into a task that was run lazily. The change is purely internal,
restructuring how/when this task is built.
diff --git a/linkerd/app/inbound/src/server.rs b/linkerd/app/inbound/src/server.rs
@@ -1,11 +1,11 @@
 use crate::{direct, Inbound};
+use futures::Stream;
 use linkerd_app_core::{
-    config::ServerConfig,
     io, profiles, serve, svc,
-    transport::{self, listen::Bind, ClientAddr, Local, OrigDstAddr, Remote, ServerAddr},
+    transport::{self, ClientAddr, Local, OrigDstAddr, Remote, ServerAddr},
     Error,
 };
-use std::{fmt::Debug, future::Future};
+use std::fmt::Debug;
 use tracing::debug_span;
 
 #[derive(Copy, Clone, Debug)]
@@ -16,72 +16,63 @@ struct TcpEndpoint {
 // === impl Inbound ===
 
 impl Inbound<()> {
-    pub fn serve<B, G, GSvc, P>(
+    pub async fn serve<A, I, G, GSvc, P>(
         self,
-        bind: B,
+        addr: Local<ServerAddr>,
+        listen: impl Stream<Item = io::Result<(A, I)>> + Send + Sync + 'static,
         profiles: P,
         gateway: G,
-    ) -> (Local<ServerAddr>, impl Future<Output = ()> + Send)
-    where
-        B: Bind<ServerConfig>,
-        B::Addrs: svc::Param<Remote<ClientAddr>>
-            + svc::Param<Local<ServerAddr>>
-            + svc::Param<OrigDstAddr>,
+    ) where
+        A: svc::Param<Remote<ClientAddr>> + svc::Param<OrigDstAddr> + Clone + Send + Sync + 'static,
+        I: io::AsyncRead + io::AsyncWrite + io::Peek + io::PeerAddr,
+        I: Debug + Unpin + Send + Sync + 'static,
         G: svc::NewService<direct::GatewayConnection, Service = GSvc>,
         G: Clone + Send + Sync + Unpin + 'static,
-        GSvc: svc::Service<direct::GatewayIo<io::ScopedIo<B::Io>>, Response = ()> + Send + 'static,
+        GSvc: svc::Service<direct::GatewayIo<io::ScopedIo<I>>, Response = ()> + Send + 'static,
         GSvc::Error: Into<Error>,
         GSvc::Future: Send,
         P: profiles::GetProfile<profiles::LookupAddr> + Clone + Send + Sync + Unpin + 'static,
         P::Error: Send,
         P::Future: Send,
     {
-        let (Local(ServerAddr(la)), listen) = bind
-            .bind(&self.config.proxy.server)
-            .expect("Failed to bind inbound listener");
+        let shutdown = self.runtime.drain.clone().signaled();
 
-        let serve = async move {
-            let shutdown = self.runtime.drain.clone().signaled();
+        // Handles connections to ports that can't be determined to be HTTP.
+        let forward = self
+            .clone()
+            .into_tcp_connect(addr.port())
+            .push_tcp_forward()
+            .into_stack()
+            .push_map_target(TcpEndpoint::from_param)
+            .instrument(|_: &_| debug_span!("tcp"))
+            .into_inner();
 
-            // Handles connections to ports that can't be determined to be HTTP.
-            let forward = self
-                .clone()
-                .into_tcp_connect(la.port())
-                .push_tcp_forward()
-                .into_stack()
-                .push_map_target(TcpEndpoint::from_param)
-                .instrument(|_: &_| debug_span!("tcp"))
-                .into_inner();
+        // Handles connections that target the inbound proxy port.
+        let direct = self
+            .clone()
+            .into_tcp_connect(addr.port())
+            .push_tcp_forward()
+            .map_stack(|_, _, s| s.push_map_target(TcpEndpoint::from_param))
+            .push_direct(gateway)
+            .into_stack()
+            .instrument(|_: &_| debug_span!("direct"))
+            .into_inner();
 
-            // Handles connections that target the inbound proxy port.
-            let direct = self
-                .clone()
-                .into_tcp_connect(la.port())
-                .push_tcp_forward()
-                .map_stack(|_, _, s| s.push_map_target(TcpEndpoint::from_param))
-                .push_direct(gateway)
-                .into_stack()
-                .instrument(|_: &_| debug_span!("direct"))
-                .into_inner();
+        // Handles HTTP connections.
+        let http = self
+            .into_tcp_connect(addr.port())
+            .push_http_router(profiles)
+            .push_http_server();
 
-            // Handles HTTP connections.
-            let http = self
-                .into_tcp_connect(la.port())
-                .push_http_router(profiles)
-                .push_http_server();
+        // Determines how to handle an inbound connection, dispatching it to the appropriate
+        // stack.
+        let server = http
+            .push_detect_http(forward.clone())
+            .push_detect_tls(forward)
+            .push_accept(addr.port(), direct)
+            .into_inner();
 
-            // Determines how to handle an inbound connection, dispatching it to the appropriate
-            // stack.
-            let server = http
-                .push_detect_http(forward.clone())
-                .push_detect_tls(forward)
-                .push_accept(la.port(), direct)
-                .into_inner();
-
-            serve::serve(listen, server, shutdown).await
-        };
-
-        (Local(ServerAddr(la)), serve)
+        serve::serve(listen, server, shutdown).await;
     }
 }
 
diff --git a/linkerd/app/outbound/src/lib.rs b/linkerd/app/outbound/src/lib.rs
@@ -16,20 +16,21 @@ pub mod tcp;
 #[cfg(test)]
 pub(crate) mod test_util;
 
+use futures::Stream;
 use linkerd_app_core::{
-    config::{ProxyConfig, ServerConfig},
-    metrics, profiles,
+    config::ProxyConfig,
+    io, metrics, profiles,
     proxy::{
         api_resolve::{ConcreteAddr, Metadata},
         core::Resolve,
     },
     serve,
     svc::{self, stack::Param},
     tls,
-    transport::{self, addrs::*, listen::Bind},
+    transport::{self, addrs::*},
     AddrMatch, Conditional, Error, ProxyRuntime,
 };
-use std::{collections::HashMap, future::Future, time::Duration};
+use std::{collections::HashMap, fmt::Debug, time::Duration};
 use tracing::info;
 
 const EWMA_DEFAULT_RTT: Duration = Duration::from_millis(30);
@@ -119,15 +120,15 @@ impl<S> Outbound<S> {
 }
 
 impl Outbound<()> {
-    pub fn serve<B, P, R>(
+    pub async fn serve<A, I, P, R>(
         self,
-        bind: B,
+        listen: impl Stream<Item = io::Result<(A, I)>> + Send + Sync + 'static,
         profiles: P,
         resolve: R,
-    ) -> (Local<ServerAddr>, impl Future<Output = ()>)
-    where
-        B: Bind<ServerConfig>,
-        B::Addrs: Param<Remote<ClientAddr>> + Param<OrigDstAddr>,
+    ) where
+        A: Param<Remote<ClientAddr>> + Param<OrigDstAddr> + Clone + Send + Sync + 'static,
+        I: io::AsyncRead + io::AsyncWrite + io::Peek + io::PeerAddr,
+        I: Debug + Unpin + Send + Sync + 'static,
         R: Clone + Send + Sync + Unpin + 'static,
         R: Resolve<ConcreteAddr, Endpoint = Metadata, Error = Error>,
         R::Resolution: Send,
@@ -136,33 +137,25 @@ impl Outbound<()> {
         P::Future: Send,
         P::Error: Send,
     {
-        let (listen_addr, listen) = bind
-            .bind(&self.config.proxy.server)
-            .expect("Failed to bind outbound listener");
-
-        let serve = async move {
-            if self.config.ingress_mode {
-                info!("Outbound routing in ingress-mode");
-                let stack = self
-                    .to_tcp_connect()
-                    .push_tcp_endpoint()
-                    .push_http_endpoint()
-                    .into_ingress(profiles, resolve);
-                let shutdown = self.runtime.drain.signaled();
-                serve::serve(listen, stack, shutdown).await;
-            } else {
-                let logical = self.to_tcp_connect().push_logical(resolve);
-                let endpoint = self.to_tcp_connect().push_endpoint();
-                let server = endpoint
-                    .push_switch_logical(logical.into_inner())
-                    .push_discover(profiles)
-                    .into_inner();
-                let shutdown = self.runtime.drain.signaled();
-                serve::serve(listen, server, shutdown).await;
-            }
-        };
-
-        (listen_addr, serve)
+        if self.config.ingress_mode {
+            info!("Outbound routing in ingress-mode");
+            let stack = self
+                .to_tcp_connect()
+                .push_tcp_endpoint()
+                .push_http_endpoint()
+                .into_ingress(profiles, resolve);
+            let shutdown = self.runtime.drain.signaled();
+            serve::serve(listen, stack, shutdown).await;
+        } else {
+            let logical = self.to_tcp_connect().push_logical(resolve);
+            let endpoint = self.to_tcp_connect().push_endpoint();
+            let server = endpoint
+                .push_switch_logical(logical.into_inner())
+                .push_discover(profiles)
+                .into_inner();
+            let shutdown = self.runtime.drain.signaled();
+            serve::serve(listen, server, shutdown).await;
+        }
     }
 }
 
diff --git a/linkerd/app/src/lib.rs b/linkerd/app/src/lib.rs
@@ -29,8 +29,7 @@ use tokio::{
     sync::mpsc,
     time::{self, Duration},
 };
-use tracing::instrument::Instrument;
-use tracing::{debug, info, info_span};
+use tracing::{debug, info, info_span, Instrument};
 
 /// Spawns a sidecar proxy.
 ///
@@ -115,8 +114,10 @@ impl Config {
 
         let dns = dns.build();
 
+        // Ensure that we've obtained a valid identity before binding any servers.
         let identity = info_span!("identity")
             .in_scope(|| identity.build(dns.resolver.clone(), metrics.control.clone()))?;
+
         let report = identity.metrics().and_then(report);
 
         let (drain_tx, drain_rx) = drain::channel();
@@ -190,14 +191,40 @@ impl Config {
             dst.resolve.clone(),
         );
 
-        let (inbound_addr, inbound_serve) =
-            inbound.serve(bind_in, dst.profiles.clone(), gateway_stack);
-        let (outbound_addr, outbound_serve) = outbound.serve(bind_out, dst.profiles, dst.resolve);
-
-        let start_proxy = Box::pin(async move {
-            tokio::spawn(outbound_serve.instrument(info_span!("outbound")));
-            tokio::spawn(inbound_serve.instrument(info_span!("inbound")));
-        });
+        // Bind the proxy sockets eagerly (so they're reserved and known) but defer building the
+        // stacks until the proxy starts running.
+        let (inbound_addr, inbound_listen) = bind_in
+            .bind(&inbound.config().proxy.server)
+            .expect("Failed to bind inbound listener");
+
+        let (outbound_addr, outbound_listen) = bind_out
+            .bind(&outbound.config().proxy.server)
+            .expect("Failed to bind outbound listener");
+
+        // Build a task that initializes and runs the proxy stacks.
+        let start_proxy = {
+            let inbound_addr = inbound_addr;
+            let profiles = dst.profiles;
+            let resolve = dst.resolve;
+
+            Box::pin(async move {
+                // TODO(ver): Block on identity.
+
+                tokio::spawn(
+                    outbound
+                        .serve(outbound_listen, profiles.clone(), resolve)
+                        .instrument(info_span!("outbound")),
+                );
+
+                // TODO(ver): Block on policy.
+
+                tokio::spawn(
+                    inbound
+                        .serve(inbound_addr, inbound_listen, profiles, gateway_stack)
+                        .instrument(info_span!("inbound")),
+                );
+            })
+        };
 
         Ok(App {
             admin,
