fuzz: allow client requests to fail

Cargo.lock

@@ -818,6 +818,7 @@ dependencies = [
  "linkerd-identity",
  "linkerd-io",
  "regex",
+ "thiserror",
  "tokio",
  "tokio-stream",
  "tokio-test",

linkerd/app/inbound/fuzz/Cargo.lock

@@ -688,6 +688,7 @@ dependencies = [
  "linkerd-identity",
  "linkerd-io",
  "regex",
+ "thiserror",
  "tokio",
  "tokio-stream",
  "tokio-test",

linkerd/app/inbound/src/http/mod.rs

@@ -324,7 +324,10 @@ pub mod fuzz_logic {
                         {
                             let rsp = http_util::http_request(&mut client, req).await;
                             tracing::info!(?rsp);
-                            let _body = http_util::body_to_string(rsp.into_body()).await;
+                            if let Ok(rsp) = rsp {
+                                let body = http_util::body_to_string(rsp.into_body()).await;
+                                tracing::info!(?body);
+                            }
                         }
                     }
                 }

linkerd/app/inbound/src/http/tests.rs

@@ -81,9 +81,9 @@ async fn unmeshed_http1_hello_world() {
         .uri("http://foo.svc.cluster.local:5550")
         .body(Body::default())
         .unwrap();
-    let rsp = http_util::http_request(&mut client, req).await;
+    let rsp = http_util::http_request(&mut client, req).await.unwrap();
     assert_eq!(rsp.status(), http::StatusCode::OK);
-    let body = http_util::body_to_string(rsp.into_body()).await;
+    let body = http_util::body_to_string(rsp.into_body()).await.unwrap();
     assert_eq!(body, "Hello world!");
 
     drop(client);
@@ -130,9 +130,9 @@ async fn downgrade_origin_form() {
         .header("l5d-orig-proto", "HTTP/1.1")
         .body(Body::default())
         .unwrap();
-    let rsp = http_util::http_request(&mut client, req).await;
+    let rsp = http_util::http_request(&mut client, req).await.unwrap();
     assert_eq!(rsp.status(), http::StatusCode::OK);
-    let body = http_util::body_to_string(rsp.into_body()).await;
+    let body = http_util::body_to_string(rsp.into_body()).await.unwrap();
     assert_eq!(body, "Hello world!");
 
     drop(client);
@@ -178,9 +178,9 @@ async fn downgrade_absolute_form() {
         .header("l5d-orig-proto", "HTTP/1.1; absolute-form")
         .body(Body::default())
         .unwrap();
-    let rsp = http_util::http_request(&mut client, req).await;
+    let rsp = http_util::http_request(&mut client, req).await.unwrap();
     assert_eq!(rsp.status(), http::StatusCode::OK);
-    let body = http_util::body_to_string(rsp.into_body()).await;
+    let body = http_util::body_to_string(rsp.into_body()).await.unwrap();
     assert_eq!(body, "Hello world!");
 
     drop(client);

linkerd/app/integration/src/client.rs

@@ -131,7 +131,7 @@ impl Client {
             res.status(),
         );
         let stream = res.into_parts().1;
-        http_util::body_to_string(stream).await
+        http_util::body_to_string(stream).await.unwrap()
     }
 
     pub fn request(

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

@@ -805,7 +805,7 @@ macro_rules! http1_tests {
             let srv = server::http1()
                 .route_async("/", |req| async move {
                     assert_eq!(req.headers()["transfer-encoding"], "chunked");
-                    let body = http_util::body_to_string(req.into_body()).await;
+                    let body = http_util::body_to_string(req.into_body()).await.unwrap();
                     assert_eq!(body, "hello");
                     Ok::<_, std::io::Error>(
                         Response::builder()
@@ -830,7 +830,7 @@ macro_rules! http1_tests {
 
             assert_eq!(resp.status(), StatusCode::OK);
             assert_eq!(resp.headers()["transfer-encoding"], "chunked");
-            let body = http_util::body_to_string(resp.into_body()).await;
+            let body = http_util::body_to_string(resp.into_body()).await.unwrap();
             assert_eq!(body, "world");
 
             // ensure panics from the server are propagated
@@ -1011,7 +1011,7 @@ macro_rules! http1_tests {
             assert_eq!(resp.status(), StatusCode::OK);
             assert_eq!(resp.headers()["content-length"], "55");
 
-            let body = http_util::body_to_string(resp.into_body()).await;
+            let body = http_util::body_to_string(resp.into_body()).await.unwrap();
 
             assert_eq!(body, "");
 
@@ -1076,7 +1076,7 @@ macro_rules! http1_tests {
                     v
                 );
 
-                let body = http_util::body_to_string(resp.into_body()).await;
+                let body = http_util::body_to_string(resp.into_body()).await.unwrap();
 
                 assert_eq!(body, "body till eof", "HTTP/{} body", v);
             }

linkerd/app/outbound/src/http/tests.rs

@@ -272,9 +272,11 @@ async fn meshed_hello_world() {
     let server = build_server(cfg, rt, profiles, resolver, connect).new_service(addrs(ep1));
     let (mut client, bg) = http_util::connect_and_accept(&mut ClientBuilder::new(), server).await;
 
-    let rsp = http_util::http_request(&mut client, Request::default()).await;
+    let rsp = http_util::http_request(&mut client, Request::default())
+        .await
+        .unwrap();
     assert_eq!(rsp.status(), http::StatusCode::OK);
-    let body = http_util::body_to_string(rsp.into_body()).await;
+    let body = http_util::body_to_string(rsp.into_body()).await.unwrap();
     assert_eq!(body, "Hello world!");
 
     drop(client);
@@ -332,9 +334,11 @@ async fn stacks_idle_out() {
 
     let server = svc.new_service(addrs(ep1));
     let (mut client, bg) = http_util::connect_and_accept(&mut ClientBuilder::new(), server).await;
-    let rsp = http_util::http_request(&mut client, Request::default()).await;
+    let rsp = http_util::http_request(&mut client, Request::default())
+        .await
+        .unwrap();
     assert_eq!(rsp.status(), http::StatusCode::OK);
-    let body = http_util::body_to_string(rsp.into_body()).await;
+    let body = http_util::body_to_string(rsp.into_body()).await.unwrap();
     assert_eq!(body, "Hello world!");
 
     drop(client);
@@ -405,11 +409,13 @@ async fn active_stacks_dont_idle_out() {
 
     let (mut client, client_bg) =
         http_util::connect_client(&mut ClientBuilder::new(), client_io).await;
-    let rsp = http_util::http_request(&mut client, Request::default()).await;
+    let rsp = http_util::http_request(&mut client, Request::default())
+        .await
+        .unwrap();
     assert_eq!(rsp.status(), http::StatusCode::OK);
     let body = http_util::body_to_string(rsp.into_body());
     let body_task = tokio::spawn(async move {
-        let body = body.await;
+        let body = body.await.unwrap();
         assert_eq!(body, "Hello world!");
     });
 
@@ -470,9 +476,11 @@ async fn unmeshed_hello_world(
     let server = build_server(cfg, rt, profiles, resolver, connect).new_service(addrs(ep1));
     let (mut client, bg) = http_util::connect_and_accept(&mut client_settings, server).await;
 
-    let rsp = http_util::http_request(&mut client, Request::default()).await;
+    let rsp = http_util::http_request(&mut client, Request::default())
+        .await
+        .unwrap();
     assert_eq!(rsp.status(), http::StatusCode::OK);
-    let body = http_util::body_to_string(rsp.into_body()).await;
+    let body = http_util::body_to_string(rsp.into_body()).await.unwrap();
     assert_eq!(body, "Hello world!");
 
     drop(client);

linkerd/app/test/Cargo.toml

@@ -32,6 +32,7 @@ tokio-stream = { version = "0.1.5", features = ["sync"] }
 tower = { version = "0.4.5", default-features = false}
 tracing = "0.1.23"
 tracing-subscriber = "0.2.16"
+thiserror = "1"
 
 [dev-dependencies.tracing-subscriber]
 version = "0.2.16"

linkerd/app/test/src/http_util.rs

@@ -1,13 +1,13 @@
 use crate::app_core::{io::BoxedIo, svc::Param, tls, Error};
 use crate::io;
+use crate::ContextError;
 use futures::FutureExt;
 use hyper::{
     body::HttpBody,
     client::conn::{Builder as ClientBuilder, SendRequest},
     Body, Request, Response,
 };
 use std::{
-    fmt,
     future::Future,
     sync::{Arc, Mutex},
 };
@@ -93,8 +93,14 @@ where
     let (client_io, proxy) = run_proxy(server).await;
     let (client, client_bg) = connect_client(client_settings, client_io).await;
     let bg = async move {
-        proxy.await.expect("proxy background task panicked")?;
-        client_bg.await.expect("client background task panicked")?;
+        proxy
+            .await
+            .expect("proxy background task panicked")
+            .map_err(ContextError::ctx("proxy background task failed"))?;
+        client_bg
+            .await
+            .expect("client background task panicked")
+            .map_err(ContextError::ctx("client background task failed"))?;
         Ok(())
     };
     (client, bg)
@@ -104,31 +110,32 @@ where
 pub async fn http_request(
     client: &mut SendRequest<Body>,
     request: Request<Body>,
-) -> Response<Body> {
+) -> Result<Response<Body>, Error> {
     let rsp = client
         .ready()
         .await
-        .expect("Client must not fail")
+        .map_err(ContextError::ctx("HTTP client poll_ready failed"))?
         .call(request)
         .await
-        .expect("Request must succeed");
+        .map_err(ContextError::ctx("HTTP client request failed"))?;
 
     tracing::info!(?rsp);
 
-    rsp
+    Ok(rsp)
 }
 
-pub async fn body_to_string<T>(body: T) -> String
+pub async fn body_to_string<T>(body: T) -> Result<String, Error>
 where
     T: HttpBody,
-    T::Error: fmt::Debug,
+    T::Error: Into<Error>,
 {
     let body = hyper::body::to_bytes(body)
         .await
-        .expect("body stream completes successfully");
-    std::str::from_utf8(&body[..])
-        .expect("body is utf-8")
-        .to_owned()
+        .map_err(ContextError::ctx("HTTP response body stream failed"))?;
+    let body = std::str::from_utf8(&body[..])
+        .map_err(ContextError::ctx("converting body to string failed"))?
+        .to_owned();
+    Ok(body)
 }
 
 impl Server {

linkerd/app/test/src/lib.rs

@@ -7,6 +7,8 @@ pub use futures::{future, FutureExt, TryFuture, TryFutureExt};
 pub use futures::stream::{Stream, StreamExt};
 pub use linkerd_app_core::{self as app_core, Addr, Error};
 pub use std::net::SocketAddr;
+use std::{borrow::Cow, fmt, panic::Location};
+use thiserror::Error;
 pub use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 pub use tokio::sync::oneshot;
 pub use tower::Service;
@@ -41,6 +43,14 @@ pub fn io() -> io::Builder {
     io::Builder::new()
 }
 
+#[derive(Error)]
+#[error("{}: {} ({}:{})", self.context, self.source, self.at.file(), self.at.line())]
+pub struct ContextError {
+    context: Cow<'static, str>,
+    source: Error,
+    at: &'static Location<'static>,
+}
+
 /// By default, disable logging in modules that are expected to error in tests.
 const DEFAULT_LOG: &str = "warn,\
                            linkerd=debug,\
@@ -68,3 +78,31 @@ pub fn trace_init() -> tracing::dispatcher::DefaultGuard {
     let (d, _) = trace_subscriber();
     tracing::dispatcher::set_default(&d)
 }
+
+// === impl ContextError ===
+
+impl ContextError {
+    #[track_caller]
+    pub(crate) fn ctx<E: Into<Error>>(context: impl Into<Cow<'static, str>>) -> impl Fn(E) -> Self {
+        let context = context.into();
+        let at = Location::caller();
+        move |error| {
+            let source = error.into();
+            tracing::error!(%source, message = %context);
+            Self {
+                context: context.clone(),
+                source,
+                at,
+            }
+        }
+    }
+}
+
+impl fmt::Debug for ContextError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // `unwrap` and `expect` panic messages always use `fmt::Debug`, so in
+        // order to get nicely formatted errors in panics, override our `Debug`
+        // impl to use `Display`.
+        fmt::Display::fmt(self, f)
+    }
+}