Dedupe discovery updates (#1759)

The proxy can receive redundant discovery updates. When this occurs, it
causes the balancer to churn, replacing an endpoint stack (and
therefore needlessly dropping a connection).

This change updates the discovery module to keep clones of the
discovered endpoint metadata, so that updated values can be compared to
eliminate duplicate updates.

Relates to linkerd/linkerd2#8677

Author: olix0r

Cargo.lock

@@ -1087,13 +1087,14 @@ dependencies = [
 name = "linkerd-proxy-discover"
 version = "0.1.0"
 dependencies = [
- "async-stream",
  "futures",
+ "indexmap",
  "linkerd-error",
  "linkerd-proxy-core",
  "linkerd-stack",
  "pin-project",
  "tokio",
+ "tokio-stream",
  "tokio-util 0.7.1",
  "tower",
  "tracing",

linkerd/app/outbound/src/endpoint.rs

@@ -14,7 +14,7 @@ use std::{
     sync::Arc,
 };
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq, Eq)]
 pub struct Endpoint<P> {
     pub addr: Remote<ServerAddr>,
     pub tls: tls::ConditionalClientTls,
@@ -141,15 +141,6 @@ impl<P> svc::Param<metrics::EndpointLabels> for Endpoint<P> {
     }
 }
 
-impl<P: std::hash::Hash> std::hash::Hash for Endpoint<P> {
-    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
-        self.addr.hash(state);
-        self.tls.hash(state);
-        self.logical_addr.hash(state);
-        self.protocol.hash(state);
-    }
-}
-
 // === EndpointFromMetadata ===
 impl FromMetadata {
     fn client_tls(metadata: &Metadata, reason: tls::NoClientTls) -> tls::ConditionalClientTls {

linkerd/proxy/api-resolve/src/metadata.rs

@@ -2,7 +2,7 @@ use http::uri::Authority;
 use linkerd_tls::client::ServerId;
 use std::collections::BTreeMap;
 
-/// Endpoint labels are lexographically ordered by key.
+/// Endpoint labels are lexigraphically ordered by key.
 pub type Labels = std::sync::Arc<BTreeMap<String, String>>;
 
 /// Metadata describing an endpoint.

linkerd/proxy/discover/Cargo.toml

@@ -12,6 +12,7 @@ Utilities to implement a Discover with the core Resolve type
 
 [dependencies]
 futures = { version = "0.3", default-features = false }
+indexmap = "1"
 linkerd-error = { path = "../../error" }
 linkerd-proxy-core = { path = "../core" }
 linkerd-stack = { path = "../../stack" }
@@ -22,6 +23,6 @@ tracing = "0.1"
 pin-project = "1"
 
 [dev-dependencies]
-async-stream = "0.3"
-tokio = { version = "1", features = ["macros"] }
-tower = { version = "0.4", default-features = false, features = ["discover", "util"] }
+tokio = { version = "1", features = ["macros", "test-util"] }
+tokio-stream = { version = "0.1", features = ["sync"] }
+tower = { version = "0.4", default-features = false, features = ["util"] }

linkerd/proxy/discover/src/from_resolve.rs

@@ -1,8 +1,9 @@
 use futures::{prelude::*, ready};
+use indexmap::{map::Entry, IndexMap};
 use linkerd_proxy_core::resolve::{Resolve, Update};
 use pin_project::pin_project;
 use std::{
-    collections::{HashSet, VecDeque},
+    collections::VecDeque,
     future::Future,
     net::SocketAddr,
     pin::Pin,
@@ -31,8 +32,14 @@ pub struct DiscoverFuture<F, E> {
 pub struct Discover<R: TryStream, E> {
     #[pin]
     resolution: R,
-    active: HashSet<SocketAddr>,
+
+    /// Changes that have been received but not yet emitted.
     pending: VecDeque<Change<SocketAddr, E>>,
+
+    /// The current state of resolved endpoints that have been observed. This is
+    /// an `IndexMap` so that the order of observed addresses is preserved
+    /// (mostly for tests).
+    active: IndexMap<SocketAddr, E>,
 }
 
 // === impl FromResolve ===
@@ -77,6 +84,7 @@ where
 {
     type Output = Result<Discover<F::Ok, E>, F::Error>;
 
+    #[inline]
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let resolution = ready!(self.project().future.try_poll(cx))?;
         Poll::Ready(Ok(Discover::new(resolution)))
@@ -89,7 +97,7 @@ impl<R: TryStream, E> Discover<R, E> {
     pub fn new(resolution: R) -> Self {
         Self {
             resolution,
-            active: HashSet::default(),
+            active: IndexMap::default(),
             pending: VecDeque::new(),
         }
     }
@@ -98,7 +106,7 @@ impl<R: TryStream, E> Discover<R, E> {
 impl<R, E> Stream for Discover<R, E>
 where
     R: TryStream<Ok = Update<E>>,
-    E: Clone + std::fmt::Debug,
+    E: Clone + Eq + std::fmt::Debug,
 {
     type Item = Result<Change<SocketAddr, E>, R::Error>;
 
@@ -111,50 +119,69 @@ where
             }
 
             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()));
-                            }
+            let update = match ready!(this.resolution.try_poll_next(cx)) {
+                Some(update) => update?,
+                None => return Poll::Ready(None),
+            };
+
+            match update {
+                Update::Reset(endpoints) => {
+                    let new_active = endpoints.into_iter().collect::<IndexMap<_, _>>();
+                    trace!(new = ?new_active, old = ?this.active, "Reset");
+
+                    for addr in this.active.keys() {
+                        // If the old addr is not in the new set, remove it.
+                        if !new_active.contains_key(addr) {
+                            trace!(%addr, "Scheduling removal");
+                            this.pending.push_back(Change::Remove(*addr));
+                        } else {
+                            trace!(%addr, "Unchanged");
                         }
-                        *this.active = active;
                     }
-                    Update::Add(endpoints) => {
-                        for (addr, endpoint) in endpoints.into_iter() {
+
+                    for (addr, endpoint) in new_active.iter() {
+                        if this.active.get(addr) != Some(endpoint) {
                             trace!(%addr, "Scheduling addition");
-                            this.active.insert(addr);
-                            this.pending.push_back(Change::Insert(addr, endpoint));
+                            this.pending
+                                .push_back(Change::Insert(*addr, endpoint.clone()));
                         }
                     }
-                    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));
+
+                    *this.active = new_active;
+                }
+
+                Update::Add(endpoints) => {
+                    for (addr, endpoint) in endpoints.into_iter() {
+                        trace!(%addr, "Scheduling addition");
+                        match this.active.entry(addr) {
+                            Entry::Vacant(entry) => {
+                                entry.insert(endpoint.clone());
+                                this.pending.push_back(Change::Insert(addr, endpoint));
+                            }
+                            Entry::Occupied(mut entry) => {
+                                if entry.get() != &endpoint {
+                                    entry.insert(endpoint.clone());
+                                    this.pending.push_back(Change::Insert(addr, endpoint));
+                                }
                             }
                         }
                     }
-                    Update::DoesNotExist => {
-                        trace!("Scheduling removals");
-                        this.pending.extend(this.active.drain().map(Change::Remove));
+                }
+
+                Update::Remove(addrs) => {
+                    for addr in addrs.into_iter() {
+                        if this.active.remove(&addr).is_some() {
+                            trace!(%addr, "Scheduling removal");
+                            this.pending.push_back(Change::Remove(addr));
+                        }
                     }
-                },
-                None => return Poll::Ready(None),
+                }
+
+                Update::DoesNotExist => {
+                    trace!("Clearing all active endpoints");
+                    this.pending
+                        .extend(this.active.drain(..).map(|(sa, _)| Change::Remove(sa)));
+                }
             }
         }
     }
@@ -163,50 +190,108 @@ where
 #[cfg(test)]
 mod tests {
     use super::Discover;
-    use async_stream::stream;
     use futures::prelude::*;
     use linkerd_error::Infallible;
     use linkerd_proxy_core::resolve::Update;
     use std::net::SocketAddr;
+    use tokio_stream::wrappers::ReceiverStream;
     use tower::discover::Change;
 
     const PORT: u16 = 8080;
     fn addr(n: u8) -> SocketAddr {
         SocketAddr::from(([10, 1, 1, n], PORT))
     }
 
-    #[tokio::test]
+    #[tokio::test(flavor = "current_thread")]
     async fn reset() {
-        tokio::pin! {
-            let stream = stream! {
-                yield Ok::<_, Infallible>(Update::Add((1..=2).map(|n| (addr(n), n)).collect()));
-                yield Ok(Update::Reset((2..=4).map(|n| (addr(n), n)).collect()));
-            };
-        }
-        let mut disco = Discover::new(stream);
-
-        for i in 1..=2 {
-            match disco.next().await.unwrap().unwrap() {
-                Change::Remove(_) => panic!("Unexpectd Remove"),
-                Change::Insert(a, n) => {
-                    assert_eq!(n, i);
-                    assert_eq!(addr(i), a);
-                }
-            }
-        }
-        match disco.next().await.unwrap().unwrap() {
-            Change::Remove(a) => assert_eq!(a, addr(1)),
-            change => panic!("Unexpected change: {:?}", change),
+        let (tx, rx) = tokio::sync::mpsc::channel(1);
+        let mut disco = Discover::new(ReceiverStream::new(rx));
+
+        // Use reset to set a new state with 3 addresses.
+        tx.try_send(Ok::<_, Infallible>(Update::Reset(
+            (1..=3).map(|n| (addr(n), n)).collect(),
+        )))
+        .expect("must send");
+        for i in 1..=3 {
+            assert!(matches!(
+                disco.try_next().await,
+                Ok(Some(Change::Insert(sa, n))) if sa == addr(i) && n == i
+            ));
         }
+
+        // Reset to a new state with 3 addresses, one of which is unchanged, one of which is
+        // changed, and one of which is added.
+        tx.try_send(Ok(Update::Reset(vec![
+            // Restore the original `2` value. We shouldn't see an update for it.
+            (addr(2), 2),
+            // Set a new value for `3`.
+            (addr(3), 4),
+            // Add a new address, too.
+            (addr(4), 5),
+        ])))
+        .expect("must send");
+        // The first address is removed now.
+        assert!(matches!(
+            disco.try_next().await,
+            Ok(Some(Change::Remove(a))) if a == addr(1)
+        ));
+        // Then process the changed and new addresses.
         for i in 3..=4 {
-            match disco.next().await.unwrap().unwrap() {
-                Change::Remove(_) => panic!("Unexpectd Remove"),
-                Change::Insert(a, n) => {
-                    assert_eq!(n, i);
-                    assert_eq!(addr(i), a);
-                }
-            }
+            assert!(matches!(
+                disco.try_next().await,
+                Ok(Some(Change::Insert(sa, n))) if sa == addr(i) && n == i + 1
+            ));
         }
+
+        // No more updates.
+        drop(tx);
         assert!(disco.next().await.is_none());
     }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn deduplicate_redundant() {
+        let (tx, rx) = tokio::sync::mpsc::channel(1);
+        let mut disco = Discover::new(ReceiverStream::new(rx));
+
+        // The initial update is observed.
+        tx.try_send(Ok::<_, Infallible>(Update::Add(vec![(addr(1), "a")])))
+            .expect("must send");
+        assert!(matches!(
+            disco.try_next().await,
+            Ok(Some(Change::Insert(sa, "a"))) if sa == addr(1)
+        ));
+
+        // A redundant update is not.
+        tx.try_send(Ok(Update::Add(vec![(addr(1), "a")])))
+            .expect("must send");
+        assert!(disco.try_next().now_or_never().is_none());
+
+        // A new value for an existing address is observed.
+        tx.try_send(Ok(Update::Add(vec![(addr(1), "b")])))
+            .expect("must send");
+        assert!(matches!(
+            disco.try_next().await,
+            Ok(Some(Change::Insert(sa, "b"))) if sa == addr(1)
+        ));
+
+        // Remove the address.
+        tx.try_send(Ok(Update::Remove(vec![addr(1)])))
+            .expect("must send");
+        assert!(matches!(
+            disco.try_next().await,
+            Ok(Some(Change::Remove(sa))) if sa == addr(1)
+        ));
+
+        // Re-adding the address is observed.
+        tx.try_send(Ok(Update::Add(vec![(addr(1), "b")])))
+            .expect("must send");
+        assert!(matches!(
+            disco.try_next().await,
+            Ok(Some(Change::Insert(sa, "b"))) if sa == addr(1)
+        ));
+
+        // No more updates.
+        drop(tx);
+        assert!(disco.next().await.is_none(),);
+    }
 }