[tonic-xds] Add the loadbalancer tower service

tonic-xds/Cargo.toml

@@ -43,6 +43,8 @@ tokio = { version = "1", features = ["sync", "time"] }
 # Used for weighted cluster selection and fractional route matching — does not need
 # cryptographic security, only statistical uniformity for traffic distribution.
 fastrand = "2"
+indexmap = "2"
+tracing = "0.1"
 tokio-stream = "0.1"
 tokio-util = "0.7"
 backoff = "0.4"

tonic-xds/src/client/endpoint.rs

@@ -170,3 +170,22 @@ pub(crate) trait Connector {
         addr: &EndpointAddress,
     ) -> crate::common::async_util::BoxFuture<Self::Service>;
 }
+
+/// Factory for creating per-cluster [`Connector`]s.
+///
+/// The implementation can use the cluster name to look up cluster-specific
+/// config (e.g., TLS settings from xDS CDS, cert providers from A29).
+///
+/// Both `Service` and `Connector` are exposed as associated types so callers
+/// can reference `MC::Service` directly without chaining through
+/// `<MC::Connector as Connector>::Service`.
+#[allow(dead_code)]
+pub(crate) trait MakeConnector: Send + Sync + 'static {
+    /// The service type produced by the connector.
+    type Service;
+    /// The connector type produced for each cluster.
+    type Connector: Connector<Service = Self::Service>;
+
+    /// Create a connector for the given cluster.
+    fn make_connector(&self, cluster_name: &str) -> std::sync::Arc<Self::Connector>;
+}

tonic-xds/src/client/loadbalance/channel_state.rs

@@ -31,6 +31,7 @@ use std::time::Duration;
 
 use pin_project_lite::pin_project;
 use tower::Service;
+use tower::load::Load;
 
 use crate::client::endpoint::{Connector, EndpointAddress};
 use crate::common::async_util::BoxFuture;
@@ -170,6 +171,14 @@ where
     }
 }
 
+impl<S: Load> Load for ReadyChannel<S> {
+    type Metric = S::Metric;
+
+    fn load(&self) -> Self::Metric {
+        self.inner.load()
+    }
+}
+
 // ---------------------------------------------------------------------------
 // EjectedChannel
 // ---------------------------------------------------------------------------

tonic-xds/src/client/loadbalance/errors.rs

@@ -0,0 +1,57 @@
+//! Errors for the load balancer.
+
+/// Errors produced by the load balancer.
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum LbError {
+    /// No ready endpoints available to serve the request.
+    #[error("no ready endpoints available")]
+    Unavailable,
+
+    /// The selected lb channel was not ready.
+    #[error("lb channel not ready: {0}")]
+    LbChannelPollReadyError(tower::BoxError),
+
+    /// The selected lb channel returned an error.
+    #[error("lb channel error: {0}")]
+    LbChannelCallError(tower::BoxError),
+
+    /// Discovery stream produced an error.
+    #[error("discovery error: {0}")]
+    DiscoverError(tower::BoxError),
+
+    /// Discovery stream is closed (returned None).
+    #[error("discovery stream is closed")]
+    DiscoverClosed,
+
+    /// Internal precondition violation (bug).
+    #[error("failed precondition: {0}")]
+    FailedPrecondition(&'static str),
+
+    /// Discovery is closed and no endpoints are connecting or ready —
+    /// no progress is possible, fail fast instead of hanging.
+    #[error("load balancer is stagnant: discovery is closed and no endpoints are available")]
+    Stagnation,
+}
+
+impl From<LbError> for tonic::Status {
+    fn from(err: LbError) -> Self {
+        match err {
+            LbError::Unavailable => tonic::Status::unavailable("no ready endpoints available"),
+            LbError::LbChannelPollReadyError(inner) => tonic::Status::unavailable(format!(
+                "error when polling readiness of lb channel: {inner}"
+            )),
+            LbError::DiscoverError(source) => {
+                tonic::Status::unavailable(format!("discovery error: {source}"))
+            }
+            LbError::DiscoverClosed => tonic::Status::unavailable("discovery stream is closed"),
+            LbError::FailedPrecondition(msg) => tonic::Status::failed_precondition(msg),
+            LbError::Stagnation => tonic::Status::unavailable(
+                "load balancer is stagnant: discovery is closed and no endpoints are available",
+            ),
+            LbError::LbChannelCallError(source) => match source.downcast::<tonic::Status>() {
+                Ok(status) => *status,
+                Err(source) => tonic::Status::unknown(format!("lb channel error: {source}")),
+            },
+        }
+    }
+}