Add trace-level logging around discovery updates (#1287)

It's helpful to have more insight into what's going on during endpoint
resolution. This change adds trace-level logging messages to various
resolver modules.

Author: olix0r

linkerd/proxy/discover/src/from_resolve.rs

@@ -9,6 +9,7 @@ use std::{
     task::{Context, Poll},
 };
 use tower::discover::Change;
+use tracing::trace;
 
 #[derive(Clone, Debug)]
 pub struct FromResolve<R, E> {
@@ -97,29 +98,36 @@ impl<R: TryStream, E> Discover<R, E> {
 impl<R, E> Stream for Discover<R, E>
 where
     R: TryStream<Ok = Update<E>>,
-    E: Clone,
+    E: Clone + std::fmt::Debug,
 {
     type Item = Result<Change<SocketAddr, E>, R::Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         loop {
             let this = self.as_mut().project();
             if let Some(change) = this.pending.pop_front() {
+                trace!(?change, "Changed");
                 return Poll::Ready(Some(Ok(change)));
             }
 
+            trace!("poll");
             match ready!(this.resolution.try_poll_next(cx)) {
                 Some(update) => match update? {
                     Update::Reset(endpoints) => {
                         let active = endpoints.iter().map(|(a, _)| *a).collect::<HashSet<_>>();
+                        trace!(new = ?active, old = ?this.active, "Reset");
                         for addr in this.active.iter() {
                             // If the old addr is not in the new set, remove it.
                             if !active.contains(addr) {
+                                trace!(%addr, "Scheduling removal");
                                 this.pending.push_back(Change::Remove(*addr));
+                            } else {
+                                trace!(%addr, "Unchanged");
                             }
                         }
                         for (addr, endpoint) in endpoints.into_iter() {
                             if !this.active.contains(&addr) {
+                                trace!(%addr, "Scheduling addition");
                                 this.pending
                                     .push_back(Change::Insert(addr, endpoint.clone()));
                             }
@@ -128,18 +136,21 @@ where
                     }
                     Update::Add(endpoints) => {
                         for (addr, endpoint) in endpoints.into_iter() {
+                            trace!(%addr, "Scheduling addition");
                             this.active.insert(addr);
                             this.pending.push_back(Change::Insert(addr, endpoint));
                         }
                     }
                     Update::Remove(addrs) => {
                         for addr in addrs.into_iter() {
                             if this.active.remove(&addr) {
+                                trace!(%addr, "Scheduling removal");
                                 this.pending.push_back(Change::Remove(addr));
                             }
                         }
                     }
                     Update::DoesNotExist => {
+                        trace!("Scheduling removals");
                         this.pending.extend(this.active.drain().map(Change::Remove));
                     }
                 },

linkerd/proxy/dns-resolve/src/lib.rs

@@ -71,6 +71,7 @@ async fn resolution(dns: dns::Resolver, na: NameAddr) -> Result<UpdateStream, Er
     //
     // Note: this can't be an async_stream, due to pinniness.
     let (addrs, expiry) = dns.resolve_addrs(na.name(), na.port()).await?;
+    debug!(?addrs, name = %na);
     let (tx, rx) = mpsc::channel(1);
     tokio::spawn(
         async move {
@@ -84,7 +85,7 @@ async fn resolution(dns: dns::Resolver, na: NameAddr) -> Result<UpdateStream, Er
             loop {
                 match dns.resolve_addrs(na.name(), na.port()).await {
                     Ok((addrs, expiry)) => {
-                        debug!(?addrs);
+                        debug!(?addrs, name = %na);
                         let eps = addrs.into_iter().map(|a| (a, ())).collect();
                         if tx.send(Ok(Update::Reset(eps))).await.is_err() {
                             trace!("Closed");

linkerd/proxy/resolve/src/map_endpoint.rs

@@ -107,6 +107,7 @@ where
     R: TryStream<Ok = resolve::Update<E>>,
     R::Error: Into<Error>,
     M: MapEndpoint<T, E>,
+    M::Out: std::fmt::Debug,
 {
     type Item = Result<resolve::Update<M::Out>, R::Error>;
 
@@ -135,6 +136,7 @@ where
             },
             None => return Poll::Ready(None),
         };
+        tracing::trace!(?update);
         Poll::Ready(Some(Ok(update)))
     }
 }

linkerd/proxy/resolve/src/recover.rs

@@ -140,6 +140,7 @@ where
     R: resolve::Resolve<T>,
     R::Future: Unpin,
     R::Resolution: Unpin,
+    R::Endpoint: std::fmt::Debug,
     E: Recover,
     E::Backoff: Unpin,
 {
@@ -155,37 +156,41 @@ where
                 State::Connected {
                     ref mut resolution,
                     ref mut is_initial,
-                } => match ready!(resolution.try_poll_next_unpin(cx)) {
-                    Some(Ok(Update::Remove(_))) if *is_initial => {
-                        debug_assert!(false, "Remove must not be initial update");
-                        tracing::debug!("Ignoring Remove after connection");
-                        // Continue polling until a useful update is received.
-                    }
-                    Some(Ok(update)) => {
-                        let update = if *is_initial {
-                            *is_initial = false;
-                            match update {
-                                Update::Add(eps) => Update::Reset(eps),
-                                up => up,
+                } => {
+                    tracing::trace!("polling");
+                    match ready!(resolution.try_poll_next_unpin(cx)) {
+                        Some(Ok(Update::Remove(_))) if *is_initial => {
+                            debug_assert!(false, "Remove must not be initial update");
+                            tracing::debug!("Ignoring Remove after connection");
+                            // Continue polling until a useful update is received.
+                        }
+                        Some(Ok(update)) => {
+                            let update = if *is_initial {
+                                *is_initial = false;
+                                match update {
+                                    Update::Add(eps) => Update::Reset(eps),
+                                    up => up,
+                                }
+                            } else {
+                                update
+                            };
+                            tracing::trace!(?update);
+                            return Poll::Ready(Some(Ok(update)));
+                        }
+                        Some(Err(e)) => {
+                            this.inner.state = State::Recover {
+                                error: Some(e.into()),
+                                backoff: None,
                             }
-                        } else {
-                            update
-                        };
-                        return Poll::Ready(Some(Ok(update)));
-                    }
-                    Some(Err(e)) => {
-                        this.inner.state = State::Recover {
-                            error: Some(e.into()),
-                            backoff: None,
                         }
-                    }
-                    None => {
-                        this.inner.state = State::Recover {
-                            error: Some(Eos(()).into()),
-                            backoff: None,
+                        None => {
+                            this.inner.state = State::Recover {
+                                error: Some(Eos(()).into()),
+                                backoff: None,
+                            }
                         }
                     }
-                },
+                }
                 _ => {}
             }