integration: improve tracing in tests (#846)

In stack tests, we set a fairly verbose `tracing` filter, since the logs
are captured by libtest and only printed if a test fails or when the
`--show-output` flag is passed. However, in integration tests, we
disable most tracing, because earlier versions of Rust did not support
output capturing for spawned threads, and the integration tests run the
proxy in a separate thread (see #771).

Now that we're on Rust 1.49, however, libtest now supports output
capturing for spawned threads, so this PR changes the integration tests
to enable more verbose logging. This should help with debugging test
failures.

I made a few other minor improvements to tracing in tests. In
particular:
* Share the *same* tracing subscriber between the test thread and the
  test proxy. This should resolve the issue where logs from the proxy
  in an integration test and logs from the test code have timestamps
  that are relative to different start times, making the logs appear
  to time travel.
* Replace some `println!`s with tracing events.

Signed-off-by: Eliza Weisman <eliza@buoyant.io>

linkerd/app/integration/src/controller.rs

@@ -392,11 +392,11 @@ where
                 tokio::spawn(cancelable(drain.clone(), f).instrument(span));
             }
         })
-        .instrument(tracing::info_span!("controller", %name, %addr)),
+        .instrument(tracing::info_span!("controller", message = %name, %addr)),
     );
 
     listening_rx.await.expect("listening_rx");
-    println!("{} listening; addr={:?}", name, addr);
+    tracing::info!(%addr, "{} listening", name);
 
     Listening {
         addr,

linkerd/app/integration/src/identity.rs

@@ -202,7 +202,7 @@ impl Controller {
     }
 
     pub async fn run(self) -> controller::Listening {
-        println!("running support identity service");
+        tracing::debug!("running support identity service");
         controller::run(
             pb::identity_server::IdentityServer::new(self),
             "support identity service",

linkerd/app/integration/src/proxy.rs

@@ -180,14 +180,6 @@ impl Listening {
 }
 
 async fn run(proxy: Proxy, mut env: TestEnv, random_ports: bool) -> Listening {
-    // Logs from spawned threads will not be captured, so use a filter that
-    // disables most of the proxy's logs by default.
-    // TODO(eliza): when we're on Rust 1.49.0+, libtest *will* capture these
-    // logs, so we can use the same default filter as other test code.
-    const DEFAULT_LOG: &str = "error,\
-                               linkerd_proxy_http=off,\
-                               linkerd_proxy_transport=off";
-
     use app::env::Strings;
 
     let controller = if let Some(controller) = proxy.controller {
@@ -272,7 +264,17 @@ async fn run(proxy: Proxy, mut env: TestEnv, random_ports: bool) -> Listening {
     }
 
     let config = app::env::parse_config(&env).unwrap();
-    let (trace, trace_handle) = super::trace_subscriber(DEFAULT_LOG);
+    let dispatch = tracing::Dispatch::default();
+    let (trace, trace_handle) = if dispatch
+        .downcast_ref::<tracing_subscriber::fmt::TestWriter>()
+        .is_some()
+    {
+        // A dispatcher was set by the test...
+        (dispatch, app_core::trace::Handle::disabled())
+    } else {
+        eprintln!("test did not set a tracing dispatcher, creating a new one for the proxy");
+        super::trace_subscriber()
+    };
 
     let (running_tx, running_rx) = oneshot::channel();
     let (term_tx, term_rx) = oneshot::channel();
@@ -348,25 +350,15 @@ async fn run(proxy: Proxy, mut env: TestEnv, random_ports: bool) -> Listening {
     let (tap_addr, identity_addr, inbound_addr, outbound_addr, metrics_addr) =
         running_rx.await.unwrap();
 
-    // printlns will show if the test fails...
-    println!(
-        "proxy running; tap={}, identity={:?}, inbound={}{}, outbound={}{}, metrics={}",
-        tap_addr
-            .as_ref()
-            .map(SocketAddr::to_string)
-            .unwrap_or_default(),
-        identity_addr,
-        inbound_addr,
-        inbound
-            .as_ref()
-            .map(|i| format!(" (SO_ORIGINAL_DST={})", i))
-            .unwrap_or_default(),
-        outbound_addr,
-        outbound
-            .as_ref()
-            .map(|o| format!(" (SO_ORIGINAL_DST={})", o))
-            .unwrap_or_default(),
-        metrics_addr,
+    tracing::info!(
+        tap.addr = ?tap_addr,
+        identity.addr = ?identity_addr,
+        inbound.addr = ?inbound_addr,
+        inbound.orig_dst = ?inbound.as_ref(),
+        outbound.addr = ?outbound_addr,
+        outbound.orig_dst = ?outbound.as_ref(),
+        metrics.addr = ?metrics_addr,
+        "proxy running",
     );
 
     Listening {

linkerd/app/integration/src/server.rs

@@ -243,12 +243,13 @@ impl Server {
                         let svc = svc.await;
                         tracing::trace!("service acquired");
                         srv_conn_count.fetch_add(1, Ordering::Release);
-                        let svc =
-                            svc.map_err(|e| println!("support/server new_service error: {}", e))?;
+                        let svc = svc.map_err(|e| {
+                            tracing::error!("support/server new_service error: {}", e)
+                        })?;
                         let result = http
                             .serve_connection(sock, svc)
                             .await
-                            .map_err(|e| println!("support/server error: {}", e));
+                            .map_err(|e| tracing::error!("support/server error: {}", e));
                         tracing::trace!(?result, "serve done");
                         result
                     };
@@ -260,8 +261,7 @@ impl Server {
 
         listening_rx.await.expect("listening_rx");
 
-        // printlns will show if the test fails...
-        println!("{:?} server running; addr={}", version, addr,);
+        tracing::info!(?version, %addr, "server running");
 
         Listening {
             addr,
@@ -320,7 +320,7 @@ impl Svc {
                 func(req)
             }
             None => {
-                println!("server 404: {:?}", req.uri().path());
+                tracing::warn!("server 404: {:?}", req.uri().path());
                 let res = http::Response::builder()
                     .status(404)
                     .body(Default::default())

linkerd/app/integration/src/tcp.rs

@@ -232,8 +232,7 @@ async fn run_server(tcp: TcpServer) -> server::Listening {
 
     started_rx.await.expect("support tcp server started");
 
-    // printlns will show if the test fails...
-    println!("tcp server (addr={}): running", addr);
+    tracing::info!(%addr, "tcp server running");
 
     server::Listening {
         addr,

linkerd/app/integration/src/tests/identity.rs

@@ -61,7 +61,7 @@ macro_rules! generate_tls_accept_test {
             .run_with_test_env(id_svc.env)
             .await;
 
-        println!("non-tls request to {}", proxy.metrics);
+        tracing::info!("non-tls request to {}", proxy.metrics);
         let non_tls_client = $make_client_non_tls(proxy.metrics, "localhost");
         assert_eventually!(
             non_tls_client
@@ -72,7 +72,7 @@ macro_rules! generate_tls_accept_test {
                 == http::StatusCode::OK
         );
 
-        println!("tls request to {}", proxy.metrics);
+        tracing::info!("tls request to {}", proxy.metrics);
         let tls_client = $make_client_tls(
             proxy.metrics,
             "localhost",

linkerd/app/integration/src/tests/transparency.rs

@@ -1379,7 +1379,7 @@ async fn http2_request_without_authority() {
         .await
         .expect("handshake error");
 
-    tokio::spawn(conn.map_err(|e| println!("conn error: {:?}", e)));
+    tokio::spawn(conn.map_err(|e| tracing::info!("conn error: {:?}", e)));
 
     let req = Request::new(hyper::Body::empty());
     // these properties are specifically what we want, and set by default

linkerd/app/test/src/lib.rs

@@ -54,11 +54,11 @@ const DEFAULT_LOG: &str = "warn,\
                            linkerd_proxy_http=error,\
                            linkerd_proxy_transport=error";
 
-pub fn trace_subscriber(default_filter: &str) -> (Dispatch, app_core::trace::Handle) {
+pub fn trace_subscriber() -> (Dispatch, app_core::trace::Handle) {
     use std::env;
     let log_level = env::var("LINKERD2_PROXY_LOG")
         .or_else(|_| env::var("RUST_LOG"))
-        .unwrap_or_else(|_| default_filter.to_owned());
+        .unwrap_or_else(|_| DEFAULT_LOG.to_string());
     let log_format = env::var("LINKERD2_PROXY_LOG_FORMAT").unwrap_or_else(|_| "PLAIN".to_string());
     env::set_var("LINKERD2_PROXY_LOG_FORMAT", &log_format);
     // This may fail, since the global log compat layer may have been
@@ -72,6 +72,6 @@ pub fn trace_subscriber(default_filter: &str) -> (Dispatch, app_core::trace::Han
 }
 
 pub fn trace_init() -> tracing::dispatcher::DefaultGuard {
-    let (d, _) = trace_subscriber(DEFAULT_LOG);
+    let (d, _) = trace_subscriber();
     tracing::dispatcher::set_default(&d)
 }

linkerd/tracing/src/lib.rs

@@ -158,6 +158,11 @@ enum Inner {
 // === impl Handle ===
 
 impl Handle {
+    /// Returns a new `handle` with tracing disabled.
+    pub fn disabled() -> Self {
+        Self(Inner::Disabled)
+    }
+
     /// Serve requests that controls the log level. The request is expected to be either a GET or PUT
     /// request. PUT requests must have a body that describes the new log level.
     pub async fn serve_level(