Prevent server errors from panicking the proxy (#943)

The TCP server currently returns an error if an `accept(2)` call fails
with an error. This error can cause the proxy to panic.

This change handles this error gracefully, emitting a warning. The
server task no longer returns a Result-type.

Author: olix0r

linkerd/app/core/src/serve.rs

@@ -15,8 +15,7 @@ pub async fn serve<M, A, I>(
     listen: impl Stream<Item = std::io::Result<(Addrs, I)>>,
     mut new_accept: M,
     shutdown: impl Future,
-) -> Result<(), Error>
-where
+) where
     I: Send + 'static,
     M: svc::NewService<Addrs, Service = A>,
     A: tower::Service<io::ScopedIo<I>, Response = ()> + Send + 'static,
@@ -27,10 +26,16 @@ where
         futures::pin_mut!(listen);
         loop {
             match listen.next().await {
-                None => return Ok(()),
+                None => return,
                 Some(conn) => {
                     // If the listener returned an error, complete the task.
-                    let (addrs, io) = conn?;
+                    let (addrs, io) = match conn {
+                        Ok(conn) => conn,
+                        Err(error) => {
+                            warn!(%error, "Server failed to accept connection");
+                            continue;
+                        }
+                    };
 
                     // The local addr should be instrumented from the listener's context.
                     let span = info_span!(
@@ -79,7 +84,7 @@ where
     // This ensures that the accept service's readiness can't block shutdown.
     tokio::select! {
         res = accept => { res }
-        _ = shutdown => { Ok(()) }
+        _ = shutdown => {}
     }
 }
 

linkerd/app/inbound/src/lib.rs

@@ -152,9 +152,7 @@ impl Inbound<()> {
                 .to_tcp_connect()
                 .into_server(listen_addr.port(), profiles, gateway);
             let shutdown = self.runtime.drain.signaled();
-            serve::serve(listen, stack, shutdown)
-                .await
-                .expect("Inbound server failed");
+            serve::serve(listen, stack, shutdown).await
         };
 
         (listen_addr, serve)

linkerd/app/outbound/src/lib.rs

@@ -187,15 +187,11 @@ impl Outbound<()> {
                     .into_inner();
                 let stack = self.to_ingress(profiles, tcp, http);
                 let shutdown = self.runtime.drain.signaled();
-                serve::serve(listen, stack, shutdown)
-                    .await
-                    .expect("Outbound server failed");
+                serve::serve(listen, stack, shutdown).await;
             } else {
                 let stack = self.to_tcp_connect().into_server(resolve, profiles);
                 let shutdown = self.runtime.drain.signaled();
-                serve::serve(listen, stack, shutdown)
-                    .await
-                    .expect("Outbound server failed");
+                serve::serve(listen, stack, shutdown).await;
             }
         };
 

linkerd/app/src/admin.rs

@@ -25,7 +25,7 @@ pub struct Config {
 pub struct Admin {
     pub listen_addr: SocketAddr,
     pub latch: admin::Latch,
-    pub serve: Pin<Box<dyn std::future::Future<Output = Result<(), Error>> + Send + 'static>>,
+    pub serve: Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>>,
 }
 
 #[derive(Debug, Default)]

linkerd/app/src/lib.rs

@@ -20,7 +20,7 @@ use linkerd_channel::into_stream::IntoStream;
 use std::{net::SocketAddr, pin::Pin};
 use tokio::{sync::mpsc, time::Duration};
 use tracing::instrument::Instrument;
-use tracing::{debug, error, info, info_span};
+use tracing::{debug, info, info_span};
 
 /// Spawns a sidecar proxy.
 ///
@@ -255,7 +255,6 @@ impl App {
                         tokio::spawn(
                             admin
                                 .serve
-                                .map_err(|e| panic!("admin server died: {}", e))
                                 .instrument(info_span!("admin", listen.addr = %admin.listen_addr)),
                         );
 
@@ -293,11 +292,7 @@ impl App {
                                     )
                                     .instrument(info_span!("tap_clean")),
                             );
-                            tokio::spawn(
-                                serve
-                                    .map_err(|error| error!(%error, "server died"))
-                                    .instrument(info_span!("tap")),
-                            );
+                            tokio::spawn(serve.instrument(info_span!("tap")));
                         }
 
                         if let oc_collector::OcCollector::Enabled(oc) = oc_collector {

linkerd/app/src/tap.rs

@@ -23,7 +23,7 @@ pub enum Tap {
     Enabled {
         listen_addr: SocketAddr,
         registry: tap::Registry,
-        serve: Pin<Box<dyn std::future::Future<Output = Result<(), Error>> + Send + 'static>>,
+        serve: Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>>,
     },
 }