Remove LocalVersions.clear() (#1278)

This changes behavior so that we won't remove all resources that were
added by a given agent from the relay Config when that agent disconnects
gracefully. This could be disrupting in the usual case where the agent
disconnects simply due to a rollout or rescheduling event.

The LocalVersion struct exits scope when the connection is closed so we
don't need to clear the local state it has either.

Since the Cluster (EndpointSet) resource is versioned deterministically
by hashing the endpoints, and updated as a batch, we don't have to worry
about old endpoints lingering if we miss an update when one agent shuts
down before another picks up.

In the case of the Datacenter resource this is handled separately
through the `client_disconnected()` method on the `Configuration` trait,
which in the case of a relay will remove the disconnecting agent from
its DatacenterMap. We also need to make sure we remove the client from
ClusterMap.localities so that the next agent can take over, so I've
added the `remove_contributor()` method that is called in
`client_disconnected()`.

Author: glindstedt

crates/xds/src/config.rs

@@ -53,22 +53,6 @@ impl LocalVersions {
             panic!("unable to retrieve `{ty}` versions, available versions are {versions:?}");
         }
     }
-
-    #[inline]
-    pub fn clear<C: crate::config::Configuration>(
-        &self,
-        config: &Arc<C>,
-        remote_addr: Option<std::net::IpAddr>,
-    ) {
-        for (type_url, map) in &self.versions {
-            let mut map = map.lock();
-            let remove = map.keys().cloned().collect::<Vec<_>>();
-            if let Err(error) = config.apply_delta(type_url, vec![], &remove, remote_addr) {
-                tracing::warn!(%error, count = remove.len(), type_url, "failed to remove resources upon connection loss");
-            }
-            map.clear();
-        }
-    }
 }
 
 pub struct ClientState {

crates/xds/src/server.rs

@@ -636,8 +636,6 @@ impl<C: crate::config::Configuration> AggregatedControlPlaneDiscoveryService for
                     tracing::info!("xds stream terminated");
                 }
 
-                local.clear(&config, Some(remote_addr));
-
                 res
             }
             .instrument(tracing::trace_span!("handle_delta_discovery_response")),

src/config.rs

@@ -335,6 +335,13 @@ impl quilkin_xds::config::Configuration for Config {
                 dc.remove(ip);
             });
         }
+        if let Some(cl) = self.dyn_cfg.clusters() {
+            cl.modify(|cl| {
+                // Make sure we remove this agent as a contributor to any locality that it has
+                // contributed endpoints for
+                cl.remove_contributor(Some(ip));
+            });
+        }
     }
 }
 

src/net/cluster.rs

@@ -547,6 +547,17 @@ where
         ret
     }
 
+    #[inline]
+    pub fn remove_contributor(&self, remote_addr: Option<std::net::IpAddr>) {
+        self.localities.retain(|k, v| {
+            let keep = *v != remote_addr;
+            if !keep {
+                tracing::debug!(locality=?k, ?remote_addr, "removing locality contributor");
+            }
+            keep
+        });
+    }
+
     #[inline]
     pub fn remove_locality(
         &self,