duplex: include src/dst peer in logs (#937)

The `linkerd_duplex` crate emits trace events for every read and write
operation performed, but these events don't make it clear which peer
performed the read or write. This branch adds names to the trace event
to make it possible to determine whether a read/write was performed by
the source or destination peer.

This does make the `Duplex` struct two words larger. Not sure if we
care. It might theoretically be possible to encode this using type
parameters to avoid the `&'static str`s; I'm not sure if this is worth
adding some wacky code to avoid two words of string constant pointer.

Signed-off-by: Eliza Weisman <[email protected]>

Author: hawkw

linkerd/duplex/src/lib.rs

@@ -23,6 +23,7 @@ struct HalfDuplex<T> {
     is_shutdown: bool,
     #[pin]
     io: T,
+    direction: &'static str,
 }
 
 /// A buffer used to copy bytes from one IO to another.
@@ -45,8 +46,8 @@ where
 {
     pub fn new(in_io: In, out_io: Out) -> Self {
         Duplex {
-            half_in: HalfDuplex::new(in_io),
-            half_out: HalfDuplex::new(out_io),
+            half_in: HalfDuplex::new(in_io, "client->server"),
+            half_out: HalfDuplex::new(out_io, "server->client"),
         }
     }
 }
@@ -78,11 +79,12 @@ impl<T> HalfDuplex<T>
 where
     T: AsyncRead + Unpin,
 {
-    fn new(io: T) -> Self {
+    fn new(io: T, direction: &'static str) -> Self {
         Self {
             buf: Some(CopyBuf::new()),
             is_shutdown: false,
             io,
+            direction,
         }
     }
 
@@ -100,14 +102,14 @@ where
         // shutdown, we finished in a previous poll, so don't even enter into
         // the copy loop.
         if dst.is_shutdown {
-            trace!("already shutdown");
+            trace!(direction = %self.direction, "already shutdown");
             return Poll::Ready(Ok(()));
         }
         loop {
             ready!(self.poll_read(cx))?;
             ready!(self.poll_write_into(dst, cx))?;
             if self.buf.is_none() {
-                trace!("shutting down");
+                trace!(direction = %self.direction, "shutting down");
                 debug_assert!(!dst.is_shutdown, "attempted to shut down destination twice");
                 ready!(Pin::new(&mut dst.io).poll_shutdown(cx))?;
                 dst.is_shutdown = true;
@@ -123,9 +125,9 @@ where
             if !buf.has_remaining() {
                 buf.reset();
 
-                trace!("reading");
+                trace!(direction = %self.direction, "reading");
                 let n = ready!(io::poll_read_buf(Pin::new(&mut self.io), cx, buf))?;
-                trace!("read {}B", n);
+                trace!(direction = %self.direction, "read {}B", n);
 
                 is_eof = n == 0;
             }
@@ -148,9 +150,9 @@ where
     {
         if let Some(ref mut buf) = self.buf {
             while buf.has_remaining() {
-                trace!("writing {}B", buf.remaining());
+                trace!(direction = %self.direction, "writing {}B", buf.remaining());
                 let n = ready!(io::poll_write_buf(Pin::new(&mut dst.io), cx, buf))?;
-                trace!("wrote {}B", n);
+                trace!(direction = %self.direction, "wrote {}B", n);
                 if n == 0 {
                     return Poll::Ready(Err(write_zero()));
                 }

linkerd/proxy/http/src/upgrade.rs

@@ -139,7 +139,7 @@ impl Drop for Inner {
             let both_upgrades = async move {
                 let (server_conn, client_conn) = tokio::try_join!(server_upgrade, client_upgrade)?;
                 trace!("HTTP upgrade successful");
-                if let Err(e) = Duplex::new(server_conn, client_conn).await {
+                if let Err(e) = Duplex::new(client_conn, server_conn).await {
                     info!("tcp duplex error: {}", e)
                 }
                 Ok::<(), ()>(())