feat(tonic-xds): implement OTEL metrics recorder (#2704)

## Motivation

Ref: #2444 

gRFC A78 defines the XdsClient OpenTelemetry metrics, and #2669 added
the framework-agnostic `MetricsRecorder` extension point in `xds-client`
— but shipped no bundled backend. Every consumer therefore had to
implement the
trait by hand just to get the metrics into OpenTelemetry. This PR
provides that backend and exposes it through `tonic-xds`.

## Solution

Now an `OtelMetricsRecorder` is bundled when `otel` feature flag is
enabled. User can provide their own OTEL meter and the five A78
XdsClient metrics (`connected`, `server_failure`,
`resource_update_valid/invalid`, `resources`) are emitted through that
meter. This PR marks the completion of metrics support in `tonic-xds`
based on the currently supported xDS features.

## Out of Scope

A few gaps exist as compared to A78, all because of other unsupported
gRFCs:

1. `nached_but_cached` and `cache_state` are not emitted, pending A88
data-error caching.
2.`grpc.xds.authority` uses `#old` sentinel for now as xDS Federation is
not yet supported.

Author: YutaoMa

Cargo.toml

@@ -18,6 +18,7 @@ members = [
   "grpc-protobuf-build",
   "protoc-gen-rust-grpc",
   "xds-client",
+  "xds-client-opentelemetry",
   "tonic-xds",
   "interop",
   "tests/disable_comments",

tonic-xds/Cargo.toml

@@ -55,6 +55,8 @@ tonic-prost = { version = "0.14", optional = true }
 rustls = { version = "0.23", default-features = false, features = ["std", "tls12"], optional = true }
 rustls-pemfile = { version = "2", optional = true }
 x509-parser = { version = "0.17", optional = true }
+opentelemetry = { version = "0.32", optional = true, default-features = false, features = ["metrics"] }
+xds-client-opentelemetry = { version = "0.1.0-alpha.1", path = "../xds-client-opentelemetry", optional = true }
 
 [lints]
 workspace = true
@@ -73,6 +75,11 @@ google-cloud-auth = { version = "1.9", default-features = false }
 [features]
 testutil = ["dep:tonic-prost"]
 
+# Bundled OpenTelemetry metrics recorder for the gRFC A78 xDS client metrics.
+# The recorder lives in the companion `xds-client-opentelemetry` crate so its
+# OpenTelemetry version is decoupled from the core crates.
+otel = ["dep:opentelemetry", "dep:xds-client-opentelemetry"]
+
 # TLS crypto backend — pick exactly one.
 _tls-any = ["dep:rustls", "dep:rustls-pemfile", "dep:x509-parser"]
 tls-ring = ["_tls-any", "tonic/tls-ring", "xds-client/tonic-tls-ring"]

tonic-xds/src/client/channel.rs

@@ -16,7 +16,9 @@ use std::sync::Arc;
 use std::task::{Context, Poll};
 use tonic::{body::Body as TonicBody, client::GrpcService, transport::channel::Channel};
 use tower::{BoxError, Service, ServiceBuilder, util::BoxCloneSyncService};
-use xds_client::{ClientConfig, Node, ProstCodec, TokioRuntime, TonicTransportBuilder, XdsClient};
+use xds_client::{
+    ClientConfig, MetricsRecorder, Node, ProstCodec, TokioRuntime, TonicTransportBuilder, XdsClient,
+};
 
 use crate::client::retry::{GrpcRetryPolicy, GrpcRetryPolicyConfig, RetryLayer};
 
@@ -162,9 +164,25 @@ const _: fn() = || {
 };
 
 /// Builder for creating an [`XdsChannel`] or [`XdsChannelGrpc`].
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct XdsChannelBuilder {
     config: Arc<XdsChannelConfig>,
+    recorder: Option<Arc<dyn MetricsRecorder>>,
+}
+
+impl Debug for XdsChannelBuilder {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("XdsChannelBuilder")
+            .field("config", &self.config)
+            .field(
+                "recorder",
+                &self
+                    .recorder
+                    .as_deref()
+                    .map_or("None", |r| std::any::type_name_of_val(r)),
+            )
+            .finish()
+    }
 }
 
 impl XdsChannelBuilder {
@@ -173,9 +191,36 @@ impl XdsChannelBuilder {
     pub fn new(config: XdsChannelConfig) -> Self {
         Self {
             config: Arc::new(config),
+            recorder: None,
         }
     }
 
+    /// Sets the [`MetricsRecorder`] backend that receives the gRFC A78 xDS
+    /// client metrics emitted by the underlying [`XdsClient`].
+    ///
+    /// By default no recorder is configured and metric emission is skipped.
+    /// With the `otel` feature, `with_otel_metrics` provides a one-call
+    /// OpenTelemetry setup.
+    #[must_use]
+    pub fn with_metrics_recorder(mut self, recorder: Arc<dyn MetricsRecorder>) -> Self {
+        self.recorder = Some(recorder);
+        self
+    }
+
+    /// Emits the gRFC A78 xDS client metrics through an OpenTelemetry `Meter`.
+    ///
+    /// Convenience wrapper over
+    /// [`with_metrics_recorder`](Self::with_metrics_recorder) that installs an
+    /// [`OtelMetricsRecorder`](xds_client_opentelemetry::OtelMetricsRecorder) from
+    /// the companion `xds-client-opentelemetry` crate. Requires the `otel` feature.
+    #[cfg(feature = "otel")]
+    #[must_use]
+    pub fn with_otel_metrics(self, meter: opentelemetry::metrics::Meter) -> Self {
+        self.with_metrics_recorder(Arc::new(
+            xds_client_opentelemetry::OtelMetricsRecorder::new(meter),
+        ))
+    }
+
     fn build_tonic_grpc_channel(&self) -> Result<XdsChannelGrpc, BuildError> {
         let bootstrap = match self.config.bootstrap.clone() {
             Some(b) => b,
@@ -214,8 +259,12 @@ impl XdsChannelBuilder {
         let node = Node::try_from(bootstrap.node)?;
         let client_config =
             ClientConfig::new(node, &server_uri).with_target(self.config.target_uri.to_string());
-        let xds_client =
-            XdsClient::builder(client_config, transport_builder, ProstCodec, TokioRuntime).build();
+        let mut client_builder =
+            XdsClient::builder(client_config, transport_builder, ProstCodec, TokioRuntime);
+        if let Some(recorder) = self.recorder.clone() {
+            client_builder = client_builder.with_metrics_recorder(recorder);
+        }
+        let xds_client = client_builder.build();
 
         let cache = Arc::new(XdsCache::new());
         let resource_manager =

tonic-xds/src/lib.rs

@@ -151,5 +151,7 @@ pub use xds::bootstrap::{BootstrapConfig, BootstrapError};
 pub use xds::uri::{XdsUri, XdsUriError};
 pub use xds_client::TonicCallCredentials;
 
+pub use xds_client::{Instrument, InstrumentKind, KeyValue, MetricsRecorder, StringValue, Value};
+
 #[cfg(any(test, feature = "testutil"))]
 pub mod testutil;

xds-client-opentelemetry/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "xds-client-opentelemetry"
+description = "OpenTelemetry metrics recorder for xds-client (gRFC A78 XdsClient metrics)"
+version = "0.1.0-alpha.1"
+edition = "2024"
+homepage = "https://github.com/grpc/grpc-rust"
+repository = "https://github.com/grpc/grpc-rust"
+license = "MIT"
+rust-version = { workspace = true }
+publish = false
+
+[lints]
+workspace = true
+
+[dependencies]
+xds-client = { version = "0.1.0-alpha.1", path = "../xds-client" }
+opentelemetry = { version = "0.32", default-features = false, features = ["metrics"] }
+
+[dev-dependencies]
+opentelemetry_sdk = { version = "0.32", default-features = false, features = ["metrics", "testing"] }
+
+[package.metadata.cargo_check_external_types]
+allowed_external_types = [
+    # The recorder accepts an OpenTelemetry `Meter` and implements the
+    # `xds-client` `MetricsRecorder` trait, so both appear in its public API.
+    "opentelemetry::*",
+    "xds_client::*",
+]