merge debug branch and try_send branch

glindstedt · glindstedt · commit dd50b2aae0c9 · 2025-10-28T13:53:34.000+01:00
diff --git a/crates/xds/src/client.rs b/crates/xds/src/client.rs
@@ -49,6 +49,8 @@ pub type MdsClient = Client<MdsGrpcClient>;
 
 pub(crate) const IDLE_REQUEST_INTERVAL: Duration = Duration::from_secs(30);
 
+const REQUEST_BUFFER_SIZE: usize = 100;
+
 #[tonic::async_trait]
 pub trait ServiceClient: Clone + Sized + Send + 'static {
     type Request: Clone + Send + Sync + Sized + 'static + std::fmt::Debug;
@@ -476,16 +478,15 @@ impl DeltaClientStream {
     ) -> Result<(Self, tonic::Streaming<DeltaDiscoveryResponse>, Endpoint)> {
         crate::metrics::actions_total(KIND_CLIENT, "connect").inc();
         if let Ok((mut client, ep)) = MdsClient::connect_with_backoff(endpoints).await {
-            let (req_tx, requests_rx) =
-                tokio::sync::mpsc::channel(100 /*ResourceType::VARIANTS.len()*/);
+            let (dcs, requests_rx) = Self::new();
 
             let receiver_stream = ReceiverStream::new(requests_rx);
             receiver_stream.publish_metric("mds_client").await;
 
             // Since we are doing exploratory requests to see if the remote endpoint supports delta streams, we unfortunately
             // need to actually send something before the full roundtrip occurs. This can be removed once delta discovery
             // is fully rolled out
-            req_tx
+            dcs.req_tx
                 .send(DeltaDiscoveryRequest {
                     node: Some(Node {
                         id: identifier.clone(),
@@ -502,22 +503,21 @@ impl DeltaClientStream {
                 .in_current_span()
                 .await
             {
-                return Ok((Self { req_tx }, stream.into_inner(), ep));
+                return Ok((dcs, stream.into_inner(), ep));
             }
         }
 
         let (mut client, ep) = AdsClient::connect_with_backoff(endpoints).await?;
 
-        let (req_tx, requests_rx) =
-            tokio::sync::mpsc::channel(100 /*ResourceType::VARIANTS.len()*/);
+        let (dcs, requests_rx) = Self::new();
 
         let receiver_stream = ReceiverStream::new(requests_rx);
         receiver_stream.publish_metric("ads_client").await;
 
         // Since we are doing exploratory requests to see if the remote endpoint supports delta streams, we unfortunately
         // need to actually send something before the full roundtrip occurs. This can be removed once delta discovery
         // is fully rolled out
-        req_tx
+        dcs.req_tx
             .send(DeltaDiscoveryRequest {
                 node: Some(Node {
                     id: identifier,
@@ -533,16 +533,16 @@ impl DeltaClientStream {
             .delta_aggregated_resources(receiver_stream)
             .in_current_span()
             .await?;
-        Ok((Self { req_tx }, stream.into_inner(), ep))
+        Ok((dcs, stream.into_inner(), ep))
     }
 
     pub(crate) fn new() -> (Self, tokio::sync::mpsc::Receiver<DeltaDiscoveryRequest>) {
-        let (req_tx, requests_rx) = tokio::sync::mpsc::channel(1);
+        let (req_tx, requests_rx) = tokio::sync::mpsc::channel(REQUEST_BUFFER_SIZE);
         (Self { req_tx }, requests_rx)
     }
 
     #[inline]
-    pub(crate) async fn refresh(
+    pub(crate) fn refresh(
         &self,
         identifier: &str,
         subs: Vec<(&'static str, Vec<String>)>,
@@ -551,33 +551,42 @@ impl DeltaClientStream {
         crate::metrics::actions_total(KIND_CLIENT, "refresh").inc();
         for (rt, names) in subs {
             let initial_resource_versions = local.get(rt).clone();
-            self.req_tx
-                .send(DeltaDiscoveryRequest {
-                    node: Some(Node {
-                        id: identifier.to_owned(),
-                        user_agent_name: "quilkin".into(),
-                        ..Node::default()
-                    }),
-                    type_url: (*rt).to_owned(),
-                    resource_names_subscribe: names.clone(),
-                    initial_resource_versions,
-                    // We (currently) never unsubscribe from resources, since we
-                    // never actually subscribe to particular ones in the first place
-                    resource_names_unsubscribe: Vec::new(),
-                    response_nonce: "".into(),
-                    error_detail: None,
-                })
-                .await?;
+            let refresh_request = DeltaDiscoveryRequest {
+                node: Some(Node {
+                    id: identifier.to_owned(),
+                    user_agent_name: "quilkin".into(),
+                    ..Node::default()
+                }),
+                type_url: (*rt).to_owned(),
+                resource_names_subscribe: names.clone(),
+                initial_resource_versions,
+                // We (currently) never unsubscribe from resources, since we
+                // never actually subscribe to particular ones in the first place
+                resource_names_unsubscribe: Vec::new(),
+                response_nonce: "".into(),
+                error_detail: None,
+            };
+            self.send_request(refresh_request)?;
         }
 
         Ok(())
     }
 
-    /// Sends an n/ack "response" in response to the remote response
+    /// Sends an n/ack "request" in response to the remote response
     #[inline]
-    pub(crate) async fn send_response(&self, response: DeltaDiscoveryRequest) -> Result<()> {
-        crate::metrics::actions_total(KIND_CLIENT, "respond").inc();
-        self.req_tx.send(response).await?;
+    pub(crate) fn ack_response(&self, ack_request: DeltaDiscoveryRequest) -> Result<()> {
+        // Save 10% or at least 10 spots of the request buffer for refresh requests
+        if self.req_tx.capacity() < (REQUEST_BUFFER_SIZE / 10).max(10) {
+            Err(eyre::eyre!("request buffer congested, dropping ACK/NACK"))
+        } else {
+            self.send_request(ack_request)
+        }
+    }
+
+    #[inline]
+    fn send_request(&self, request: DeltaDiscoveryRequest) -> Result<()> {
+        crate::metrics::actions_total(KIND_CLIENT, "send_request").inc();
+        self.req_tx.try_send(request)?;
         Ok(())
     }
 }
@@ -593,7 +602,7 @@ impl DeltaServerStream {
         identifier: String,
     ) -> Result<(Self, tonic::Streaming<DeltaDiscoveryRequest>)> {
         crate::metrics::actions_total(KIND_SERVER, "connect").inc();
-        let (res_tx, responses_rx) = tokio::sync::mpsc::channel(100);
+        let (res_tx, responses_rx) = tokio::sync::mpsc::channel(REQUEST_BUFFER_SIZE);
 
         res_tx
             .send(DeltaDiscoveryResponse {
@@ -630,19 +639,16 @@ pub async fn delta_subscribe<C: crate::config::Configuration>(
     notifier: Option<tokio::sync::mpsc::UnboundedSender<String>>,
     resources: &'static [(&'static str, &'static [(&'static str, Vec<String>)])],
 ) -> eyre::Result<tokio::task::JoinHandle<Result<()>>> {
-    let (mut ds, mut stream, mut connected_endpoint) = match DeltaClientStream::connect(
-        &endpoints,
-        identifier.clone(),
-    )
-    .await
-    {
-        Ok(ds) => ds,
-        Err(err) => {
-            crate::metrics::errors_total(KIND_CLIENT, "connect").inc();
-            tracing::error!(error = ?err, "failed to acquire aggregated delta stream from management server");
-            return Err(err);
-        }
-    };
+    let (mut client, mut response_stream, mut connected_endpoint) =
+        DeltaClientStream::connect(&endpoints, identifier.clone())
+            .await
+            .inspect_err(|error| {
+                crate::metrics::errors_total(KIND_CLIENT, "connect").inc();
+                tracing::error!(
+                    ?error,
+                    "failed to acquire aggregated delta stream from management server"
+                );
+            })?;
 
     async fn handle_first_response(
         stream: &mut tonic::Streaming<DeltaDiscoveryResponse>,
@@ -688,7 +694,7 @@ pub async fn delta_subscribe<C: crate::config::Configuration>(
     }
 
     let (mut control_plane, resource_subscriptions) = match handle_first_response(
-        &mut stream,
+        &mut response_stream,
         resources,
     )
     .await
@@ -708,10 +714,7 @@ pub async fn delta_subscribe<C: crate::config::Configuration>(
     let local = Arc::new(crate::config::LocalVersions::new(
         resource_subscriptions.iter().map(|(s, _)| *s),
     ));
-    if let Err(err) = ds
-        .refresh(&identifier, resource_subscriptions.to_vec(), &local)
-        .await
-    {
+    if let Err(err) = client.refresh(&identifier, resource_subscriptions.to_vec(), &local) {
         crate::metrics::errors_total(KIND_CLIENT, "request_failed").inc();
         tracing::error!(error = ?err, "failed to send initial resource requests");
         return Err(err);
@@ -721,15 +724,15 @@ pub async fn delta_subscribe<C: crate::config::Configuration>(
     let handle = tokio::task::spawn(
         async move {
             tracing::trace!("starting xDS delta stream task");
-            let mut stream = stream;
+            let mut response_stream = response_stream;
             let mut resource_subscriptions = resource_subscriptions;
 
             is_healthy.store(true, Ordering::SeqCst);
             loop {
                 tracing::info!(%control_plane, "creating discovery response handler");
-                let mut response_stream = crate::config::handle_delta_discovery_responses(
+                let mut ack_request_stream = crate::config::handle_delta_discovery_responses(
                     control_plane.clone(),
-                    stream,
+                    response_stream,
                     config.clone(),
                     local.clone(),
                     None,
@@ -738,18 +741,18 @@ pub async fn delta_subscribe<C: crate::config::Configuration>(
 
                 tracing::info!(%control_plane, "entering xDS stream loop");
                 loop {
-                    let next_response =
-                        tokio::time::timeout(IDLE_REQUEST_INTERVAL, response_stream.next());
+                    let next_ack_request =
+                        tokio::time::timeout(IDLE_REQUEST_INTERVAL, ack_request_stream.next());
 
-                    match next_response.await {
-                        Ok(Some(Ok(response))) => {
-                            let node_id = if let Some(node) = &response.node {
+                    match next_ack_request.await {
+                        Ok(Some(Ok(ack_request))) => {
+                            let node_id = if let Some(node) = &ack_request.node {
                                 node.id.clone()
                             } else {
                                 "unknown".into()
                             };
                             tracing::trace!(%node_id, "received delta response");
-                            if let Err(error) = ds.send_response(response).await {
+                            if let Err(error) = client.ack_response(ack_request) {
                                 crate::metrics::errors_total(KIND_CLIENT, "ack_failed").inc();
                                 tracing::error!(%error, %node_id, "failed to ack delta response");
                             }
@@ -777,12 +780,11 @@ pub async fn delta_subscribe<C: crate::config::Configuration>(
                         }
                         Err(_) => {
                             tracing::debug!("exceeded idle request interval sending new requests");
-                            if let Err(error) = ds
-                                .refresh(&identifier, resource_subscriptions.to_vec(), &local)
-                                .await
+                            if let Err(error) =
+                                client.refresh(&identifier, resource_subscriptions.to_vec(), &local)
                             {
                                 crate::metrics::errors_total(KIND_CLIENT, "refresh").inc();
-                                return Err(error.wrap_err("refresh failed"));
+                                tracing::error!(?error, "refresh failed");
                             }
                         }
                     }
@@ -794,7 +796,7 @@ pub async fn delta_subscribe<C: crate::config::Configuration>(
                     tracing::info!(%control_plane, "Lost connection to xDS, retrying");
                     match DeltaClientStream::connect(&endpoints, identifier.clone()).await {
                         Ok(res) => {
-                            (ds, stream, connected_endpoint) = res;
+                            (client, response_stream, connected_endpoint) = res;
                         }
                         Err(error) => {
                             crate::metrics::errors_total(KIND_CLIENT, "connect").inc();
@@ -803,7 +805,7 @@ pub async fn delta_subscribe<C: crate::config::Configuration>(
                         }
                     }
 
-                    match handle_first_response(&mut stream, resources).await {
+                    match handle_first_response(&mut response_stream, resources).await {
                         Ok((id, rs)) => {
                             control_plane = id;
                             resource_subscriptions = rs;
@@ -816,9 +818,8 @@ pub async fn delta_subscribe<C: crate::config::Configuration>(
                     }
                 }
 
-                if let Err(error) = ds
-                    .refresh(&identifier, resource_subscriptions.to_vec(), &local)
-                    .await
+                if let Err(error) =
+                    client.refresh(&identifier, resource_subscriptions.to_vec(), &local)
                 {
                     crate::metrics::errors_total(KIND_CLIENT, "refresh").inc();
                     return Err(error.wrap_err("refresh failed"));
diff --git a/crates/xds/src/metrics.rs b/crates/xds/src/metrics.rs
@@ -17,7 +17,7 @@
 use once_cell::sync::Lazy;
 use prometheus::{IntCounterVec, IntGaugeVec, Registry};
 
-pub(crate) const NODE_LABEL: &str = "node";
+pub(crate) const CLIENT_ID_LABEL: &str = "client_id";
 pub(crate) const CONTROL_PLANE_LABEL: &str = "control_plane";
 pub(crate) const TYPE_LABEL: &str = "type";
 pub(crate) const KIND_CLIENT: &str = "client";
@@ -57,20 +57,20 @@ pub(crate) fn active_control_planes(control_plane: &str) -> prometheus::IntGauge
     ACTIVE_CONTROL_PLANES.with_label_values(&[control_plane])
 }
 
-pub(crate) fn delta_discovery_requests(node: &str, type_url: &str) -> prometheus::IntCounter {
+pub(crate) fn delta_discovery_requests(client_id: &str, type_url: &str) -> prometheus::IntCounter {
     static DELTA_DISCOVERY_REQUESTS: Lazy<IntCounterVec> = Lazy::new(|| {
         prometheus::register_int_counter_vec_with_registry! {
             prometheus::opts! {
                 "delta_discovery_requests",
                 "Total number of xDS delta discovery requests",
             },
-            &[NODE_LABEL, TYPE_LABEL],
+            &[CLIENT_ID_LABEL, TYPE_LABEL],
             crate::metrics::registry(),
         }
         .unwrap()
     });
 
-    DELTA_DISCOVERY_REQUESTS.with_label_values(&[node, type_url])
+    DELTA_DISCOVERY_REQUESTS.with_label_values(&[client_id, type_url])
 }
 
 pub(crate) fn delta_discovery_responses(
diff --git a/crates/xds/src/server.rs b/crates/xds/src/server.rs
@@ -628,7 +628,6 @@ impl<C: crate::config::Configuration> AggregatedControlPlaneDiscoveryService for
                     config.interested_resources(&server_version).collect(),
                     &local,
                 )
-                .await
                 .map_err(|error| tonic::Status::internal(error.to_string()))?;
 
                 let mut response_stream = crate::config::handle_delta_discovery_responses(
@@ -649,7 +648,8 @@ impl<C: crate::config::Configuration> AggregatedControlPlaneDiscoveryService for
                             match nr {
                                 Ok(Some(Ok(ack))) => {
                                     tracing::trace!("sending ack request");
-                                    ds.send_response(ack).await.map_err(|_err| {
+                                    ds.ack_response(ack).map_err(|_err| {
+                                        crate::metrics::errors_total(KIND_SERVER, "ack_failed").inc();
                                         tonic::Status::internal("this should not be reachable")
                                     })?;
                                 }
@@ -670,7 +670,6 @@ impl<C: crate::config::Configuration> AggregatedControlPlaneDiscoveryService for
                                         config.interested_resources(&server_version).collect(),
                                         &local,
                                     )
-                                    .await
                                     .map_err(|error| tonic::Status::internal(error.to_string()))?;
                                 }
                             }

Original file line number	Diff line number	Diff line change
`@@ -628,7 +628,6 @@ impl<C: crate::config::Configuration> AggregatedControlPlaneDiscoveryService for`
`628`	`628`	`config.interested_resources(&server_version).collect(),`
`629`	`629`	`&local,`
`630`	`630`	`)`
`631`		`- .await`
`632`	`631`	`.map_err(\|error\| tonic::Status::internal(error.to_string()))?;`
`633`	`632`
`634`	`633`	`let mut response_stream = crate::config::handle_delta_discovery_responses(`
`@@ -649,7 +648,8 @@ impl<C: crate::config::Configuration> AggregatedControlPlaneDiscoveryService for`
`649`	`648`	`match nr {`
`650`	`649`	`Ok(Some(Ok(ack))) => {`
`651`	`650`	`tracing::trace!("sending ack request");`
`652`		`- ds.send_response(ack).await.map_err(\|_err\| {`
	`651`	`+ ds.ack_response(ack).map_err(\|_err\| {`
	`652`	`+ crate::metrics::errors_total(KIND_SERVER, "ack_failed").inc();`
`653`	`653`	`tonic::Status::internal("this should not be reachable")`
`654`	`654`	`})?;`
`655`	`655`	`}`
`@@ -670,7 +670,6 @@ impl<C: crate::config::Configuration> AggregatedControlPlaneDiscoveryService for`
`670`	`670`	`config.interested_resources(&server_version).collect(),`
`671`	`671`	`&local,`
`672`	`672`	`)`
`673`		`- .await`
`674`	`673`	`.map_err(\|error\| tonic::Status::internal(error.to_string()))?;`
`675`	`674`	`}`
`676`	`675`	`}`