Adopt tokio-tracing contexts (#374)

We had our own poorly-implemented logging contexts. This change
replaces all uses of the old context implementation with `tracing`
(usually via `tracing-futures`).

Author: olix0r

Cargo.lock

@@ -646,7 +646,7 @@ dependencies = [
  "tower-spawn-ready 0.1.0 (git+https://github.com/tower-rs/tower)",
  "tower-util 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tracing 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)",
- "tracing-futures 0.0.1-alpha.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tracing-futures 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tracing-log 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tracing-subscriber 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "trust-dns-resolver 0.10.2 (git+https://github.com/bluejekyll/trust-dns?rev=7c8a0739dad495bf5a4fddfe86b8bbe2aa52d060)",
@@ -711,7 +711,7 @@ dependencies = [
  "tower 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "tower-util 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tracing 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)",
- "tracing-futures 0.0.1-alpha.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tracing-futures 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -1916,7 +1916,7 @@ dependencies = [
 
 [[package]]
 name = "tracing-futures"
-version = "0.0.1-alpha.1"
+version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "futures 0.1.26 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -2402,7 +2402,7 @@ dependencies = [
 "checksum tracing 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)" = "c21ff9457accc293386c20e8f754d0b059e67e325edf2284f04230d125d7e5ff"
 "checksum tracing-attributes 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "5e27d1065a1de5d8ad2637e41fe14d3cd14363d4a20cb99090b9012004955637"
 "checksum tracing-core 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "528c8ebaaa16cdac34795180b046c031775b0d56402704d98c096788f33d646a"
-"checksum tracing-futures 0.0.1-alpha.1 (registry+https://github.com/rust-lang/crates.io-index)" = "08c7446f4fb35df7ba2c537b7e2f812f91b20a58aa2b846f028342c4d2429be0"
+"checksum tracing-futures 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "4d73cd4d483f6a9af4f9f6da37c1c162e1f99017ba708a3042e6e90e54d26e5c"
 "checksum tracing-log 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "652bc99e1286541d6ccc42d5fb37213d1cdde544f88b19fac3d94e3117b55163"
 "checksum tracing-subscriber 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)" = "c63b1771a75314374703309107e685805628c04643e6dd2c7a5cf8f94348c62e"
 "checksum trust-dns-proto 0.6.0 (git+https://github.com/bluejekyll/trust-dns?rev=7c8a0739dad495bf5a4fddfe86b8bbe2aa52d060)" = "<none>"

Cargo.toml

@@ -105,7 +105,7 @@ trust-dns-resolver = { git = "https://github.com/bluejekyll/trust-dns", rev = "7
 
 # tracing
 tracing            = "0.1.9"
-tracing-futures    = "0.0.1-alpha.1"
+tracing-futures    = "0.1"
 tracing-log        = "0.1"
 
 # tls

lib/linkerd2-proxy-core/src/listen.rs

@@ -9,14 +9,18 @@ pub trait Listen {
 
     fn poll_accept(&mut self) -> Poll<Self::Connection, Self::Error>;
 
-    fn serve<A: Accept<Self::Connection>>(self, accept: A) -> Serve<Self, A>
+    fn serve<A: Accept<Self::Connection>>(
+        self,
+        accept: A,
+    ) -> Serve<Self, A, tokio::executor::DefaultExecutor>
     where
         Self: Sized,
-        Serve<Self, A>: Future,
+        Serve<Self, A, tokio::executor::DefaultExecutor>: Future,
     {
         Serve {
             listen: self,
             accept,
+            executor: tokio::executor::DefaultExecutor::current(),
         }
     }
 }
@@ -50,26 +54,43 @@ where
     }
 }
 
-pub struct Serve<L, A> {
+pub struct Serve<L, A, E> {
     listen: L,
     accept: A,
+    executor: E,
 }
 
-impl<L, A> Future for Serve<L, A>
+impl<L, A> Serve<L, A, tokio::executor::DefaultExecutor>
 where
     L: Listen,
     A: Accept<L::Connection, Error = Never>,
     A::Future: Send + 'static,
+{
+    pub fn with_executor<E: tokio::executor::Executor>(self, executor: E) -> Serve<L, A, E> {
+        Serve {
+            listen: self.listen,
+            accept: self.accept,
+            executor,
+        }
+    }
+}
+
+impl<L, A, E> Future for Serve<L, A, E>
+where
+    L: Listen,
+    A: Accept<L::Connection, Error = Never>,
+    A::Future: Send + 'static,
+    E: tokio::executor::Executor,
 {
     type Item = Never;
-    type Error = L::Error;
+    type Error = Error;
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         loop {
-            try_ready!(self.accept.poll_ready().map_err(|e| match e {}));
-            let conn = try_ready!(self.listen.poll_accept());
-
-            tokio::spawn(self.accept.accept(conn).map_err(|e| match e {}));
+            try_ready!(self.accept.poll_ready());
+            let conn = try_ready!(self.listen.poll_accept().map_err(Into::into));
+            let accept = self.accept.accept(conn).map_err(|e| match e {});
+            self.executor.spawn(Box::new(accept)).map_err(Error::from)?;
         }
     }
 }

lib/linkerd2-proxy-discover/Cargo.toml

@@ -14,7 +14,7 @@ indexmap = "1.0"
 tokio = "0.1"
 tower = "0.1"
 tracing = "0.1"
-tracing-futures = "0.0.1-alpha.1"
+tracing-futures = "0.1"
 
 [dev-dependencies]
 tower-util = "0.1"

lib/linkerd2-proxy-transport/src/lib.rs

@@ -8,13 +8,15 @@ mod listen;
 pub mod metrics;
 mod peek;
 mod prefixed;
+mod source;
 pub mod tls;
 
 pub use self::{
     addr_info::{AddrInfo, GetOriginalDst, SoOriginalDst},
     io::BoxedIo,
     listen::Listen,
     peek::Peek,
+    source::Source,
     tls::Connection,
 };
 

lib/linkerd2-proxy-transport/src/source.rs

@@ -0,0 +1,42 @@
+use super::tls::PeerIdentity;
+use std::net::SocketAddr;
+
+/// Describes an accepted connection.
+#[derive(Clone, Debug)]
+pub struct Source {
+    pub remote: SocketAddr,
+    pub local: SocketAddr,
+    pub orig_dst: Option<SocketAddr>,
+    pub tls_peer: PeerIdentity,
+}
+
+impl Source {
+    pub fn orig_dst_if_not_local(&self) -> Option<SocketAddr> {
+        self.orig_dst.and_then(|orig_dst| {
+            // If the original destination is actually the listening socket,
+            // we don't want to create a loop.
+            if Self::same_addr(orig_dst, self.local) {
+                None
+            } else {
+                Some(orig_dst)
+            }
+        })
+    }
+
+    fn same_addr(a0: SocketAddr, a1: SocketAddr) -> bool {
+        use std::net::IpAddr::{V4, V6};
+        (a0.port() == a1.port())
+            && match (a0.ip(), a1.ip()) {
+                (V6(a0), V4(a1)) => a0.to_ipv4() == Some(a1),
+                (V4(a0), V6(a1)) => Some(a0) == a1.to_ipv4(),
+                (a0, a1) => (a0 == a1),
+            }
+    }
+}
+
+// for logging context
+impl std::fmt::Display for Source {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.remote.fmt(f)
+    }
+}

src/app/config.rs

@@ -1,6 +1,7 @@
 use super::control::ControlAddr;
 use super::identity;
 use crate::addr::{self, Addr};
+pub use crate::proxy::http::h2::Settings as H2Settings;
 use crate::transport::tls;
 use crate::{dns, Conditional};
 use indexmap::IndexSet;
@@ -143,12 +144,6 @@ pub struct Config {
     pub h2_settings: H2Settings,
 }
 
-#[derive(Copy, Clone, Debug, Default)]
-pub struct H2Settings {
-    pub initial_stream_window_size: Option<u32>,
-    pub initial_connection_window_size: Option<u32>,
-}
-
 /// Configuration settings for the tap server
 #[derive(Debug)]
 pub struct TapSettings {

src/app/control.rs

@@ -1,10 +1,10 @@
-use crate::{transport::tls, Addr};
+use linkerd2_addr::Addr;
 use std::fmt;
 
 #[derive(Clone, Debug)]
 pub struct ControlAddr {
     pub addr: Addr,
-    pub identity: tls::PeerIdentity,
+    pub identity: crate::transport::tls::PeerIdentity,
 }
 
 impl fmt::Display for ControlAddr {
@@ -16,9 +16,10 @@ impl fmt::Display for ControlAddr {
 /// Sets the request's URI from `Config`.
 pub mod add_origin {
     use super::ControlAddr;
-    use crate::{svc, Error};
+    use crate::svc;
     use futures::try_ready;
     use futures::{Future, Poll};
+    use linkerd2_error::Error;
     use std::marker::PhantomData;
     use tower_request_modifier::{Builder, RequestModifier};
 
@@ -145,10 +146,14 @@ pub mod add_origin {
 /// Resolves the controller's `addr` once before building a client.
 pub mod resolve {
     use super::{client, ControlAddr};
-    use crate::{dns, logging, svc, Addr};
+    use crate::dns;
+    use crate::svc;
     use futures::{try_ready, Future, Poll};
+    use linkerd2_addr::Addr;
     use std::net::SocketAddr;
     use std::{error, fmt};
+    use tracing::info_span;
+    use tracing_futures::{Instrument, Instrumented};
 
     #[derive(Clone, Debug)]
     pub struct Layer {
@@ -177,7 +182,7 @@ pub mod resolve {
             config: ControlAddr,
             stack: M,
         },
-        Inner(M::Future),
+        Inner(Instrumented<M::Future>),
     }
 
     #[derive(Debug)]
@@ -263,10 +268,10 @@ pub mod resolve {
             let target = client::Target {
                 addr,
                 server_name: dst.identity.clone(),
-                log_ctx: logging::admin().client("control", dst.addr.clone()),
             };
 
-            State::Inner(mk_svc.call(target))
+            info_span!("control", peer.addr = %addr, peer.identity = ?dst.identity)
+                .in_scope(move || State::Inner(mk_svc.call(target).in_current_span()))
         }
     }
 
@@ -289,15 +294,14 @@ pub mod resolve {
 pub mod client {
     use super::super::config::H2Settings;
     use crate::transport::{connect, tls};
-    use crate::{logging, proxy::http, svc, task, Addr};
+    use crate::{proxy::http, svc};
     use futures::Poll;
     use std::net::SocketAddr;
 
     #[derive(Clone, Debug)]
     pub struct Target {
         pub(super) addr: SocketAddr,
         pub(super) server_name: tls::PeerIdentity,
-        pub(super) log_ctx: logging::Client<&'static str, Addr>,
     }
 
     #[derive(Debug)]
@@ -326,7 +330,7 @@ pub mod client {
         http::h2::Connect<C, B>: svc::Service<Target>,
     {
         svc::layer::mk(|mk_conn| {
-            let inner = http::h2::Connect::new(mk_conn, task::LazyExecutor, H2Settings::default());
+            let inner = http::h2::Connect::new(mk_conn, H2Settings::default());
             Client { inner }
         })
     }
@@ -341,13 +345,13 @@ pub mod client {
         type Error = <http::h2::Connect<C, B> as svc::Service<Target>>::Error;
         type Future = <http::h2::Connect<C, B> as svc::Service<Target>>::Future;
 
+        #[inline]
         fn poll_ready(&mut self) -> Poll<(), Self::Error> {
             self.inner.poll_ready()
         }
 
+        #[inline]
         fn call(&mut self, target: Target) -> Self::Future {
-            let exe = target.log_ctx.clone().with_remote(target.addr).executor();
-            self.inner.set_executor(exe);
             self.inner.call(target)
         }
     }

src/app/errors.rs

@@ -1,8 +1,10 @@
 //! Layer to map HTTP service errors into appropriate `http::Response`s.
 
-use crate::{proxy::http::HasH2Reason, svc, Error};
+use crate::proxy::http::HasH2Reason;
+use crate::svc;
 use futures::{Future, Poll};
 use http::{header, Request, Response, StatusCode, Version};
+use linkerd2_error::Error;
 use tracing::{debug, error, warn};
 
 /// Layer to map HTTP service errors into appropriate `http::Response`s.
@@ -27,6 +29,12 @@ pub struct ResponseFuture<F> {
     is_http2: bool,
 }
 
+#[derive(Clone, Debug)]
+pub struct StatusError {
+    pub status: http::StatusCode,
+    pub message: String,
+}
+
 impl<M> svc::Layer<M> for Layer {
     type Service = Stack<M>;
 
@@ -107,7 +115,6 @@ where
 }
 
 fn map_err_to_5xx(e: Error) -> StatusCode {
-    use crate::app::outbound;
     use crate::proxy::buffer;
     use crate::proxy::http::router::error as router;
     use tower::load_shed::error as shed;
@@ -124,12 +131,20 @@ fn map_err_to_5xx(e: Error) -> StatusCode {
     } else if let Some(_) = e.downcast_ref::<router::NotRecognized>() {
         error!("could not recognize request");
         http::StatusCode::BAD_GATEWAY
-    } else if let Some(err) = e.downcast_ref::<outbound::RequireIdentityError>() {
-        error!("{}", err);
-        http::StatusCode::FORBIDDEN
+    } else if let Some(err) = e.downcast_ref::<StatusError>() {
+        error!(%err.status, %err.message);
+        err.status
     } else {
         // we probably should have handled this before?
         error!("unexpected error: {}", e);
         http::StatusCode::BAD_GATEWAY
     }
 }
+
+impl std::fmt::Display for StatusError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.message.fmt(f)
+    }
+}
+
+impl std::error::Error for StatusError {}