grpc: add ConnectionInfo to client's ResponseHeaders and Trailers

grpc/src/client/channel.rs

@@ -361,10 +361,10 @@ impl Invoke for Arc<ActiveChannel> {
         let mut i = self.lb_watcher.iter();
         loop {
             let Some(state) = i.next().await else {
-                return FailingRecvStream::new_stream_pair(StatusError::new(
-                    StatusCodeError::Internal,
-                    "channel has been closed",
-                ));
+                return FailingRecvStream::new_stream_pair(
+                    StatusError::new(StatusCodeError::Internal, "channel has been closed"),
+                    None,
+                );
             };
             let result = &state.picker.pick(&headers);
             match result {
@@ -381,7 +381,7 @@ impl Invoke for Arc<ActiveChannel> {
                     // Continue and retry the RPC with the next picker.
                 }
                 PickResult::Fail(status) => {
-                    return FailingRecvStream::new_stream_pair(status.clone());
+                    return FailingRecvStream::new_stream_pair(status.clone(), None);
                 }
                 PickResult::Drop(status) => {
                     todo!("dropped pick: {:?}", status);

grpc/src/client/interceptor.rs

@@ -563,7 +563,9 @@ mod test {
             .unwrap();
         assert_eq!(controller.recv_req().await.0, one);
         controller
-            .send_resp(ResponseStreamItem::Headers(ResponseHeaders::default()))
+            .send_resp(ResponseStreamItem::Headers(ResponseHeaders::new(
+                crate::core::test_peer_info(),
+            )))
             .await;
 
         let resp = rx.recv(&mut ByteRecvMsg::new()).await;

grpc/src/client/metadata_utils.rs

@@ -250,7 +250,7 @@ mod tests {
         // Send a Headers response on the call.
         let mut resp_md = MetadataMap::new();
         resp_md.insert("x-resp-header", "resp-value".parse().unwrap());
-        let mut headers = ResponseHeaders::default();
+        let mut headers = ResponseHeaders::new(crate::core::test_peer_info());
         *headers.metadata_mut() = resp_md;
         controller
             .send_resp(ResponseStreamItem::Headers(headers))

grpc/src/client/mod.rs

@@ -53,6 +53,7 @@ use std::time::Instant;
 
 use tonic::async_trait;
 
+use crate::core::PeerInfo;
 use crate::core::RecvMessage;
 use crate::core::SendMessage;
 use crate::metadata::MetadataMap;
@@ -367,15 +368,19 @@ impl<'a> RecvStream for Box<dyn DynRecvStream + 'a> {
 }
 
 /// Contains all information transmitted in the response headers of an RPC.
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone)]
 pub struct ResponseHeaders {
     metadata: MetadataMap,
+    peer_info: PeerInfo,
 }
 
 impl ResponseHeaders {
     /// Returns a default ResponseHeaders instance.
-    pub fn new() -> Self {
-        Self::default()
+    pub fn new(peer_info: PeerInfo) -> Self {
+        Self {
+            metadata: MetadataMap::default(),
+            peer_info,
+        }
     }
 
     /// Replaces the metadata of self with `metadata`.
@@ -397,6 +402,17 @@ impl ResponseHeaders {
     pub(crate) fn into_metadata(self) -> MetadataMap {
         self.metadata
     }
+
+    /// Replaces the peer_info of self with `peer_info`.
+    pub fn with_peer_info(mut self, peer_info: PeerInfo) -> Self {
+        self.peer_info = peer_info;
+        self
+    }
+
+    /// Replaces the peer_info of self with `peer_info`.
+    pub fn peer_info(&self) -> &PeerInfo {
+        &self.peer_info
+    }
 }
 
 /// Contains all information transmitted in the request headers of an RPC.
@@ -427,7 +443,7 @@ impl RequestHeaders {
     }
 
     /// Returns the full (e.g. "/Service/Method") method name for these headers.
-    pub fn method_name(&self) -> &str {
+    pub fn method_name(&self) -> &String {
         &self.method_name
     }
 
@@ -454,6 +470,7 @@ impl RequestHeaders {
 pub struct Trailers {
     status: crate::Result<()>,
     metadata: MetadataMap,
+    peer_info: Option<PeerInfo>,
 }
 
 impl Trailers {
@@ -462,6 +479,7 @@ impl Trailers {
         Self {
             status,
             metadata: MetadataMap::default(),
+            peer_info: None,
         }
     }
 
@@ -497,6 +515,23 @@ impl Trailers {
         self.status
     }
 
+    /// Replaces the peer info in self with `peer_info`.
+    pub fn with_peer_info(mut self, peer_info: Option<PeerInfo>) -> Self {
+        self.peer_info = peer_info;
+        self
+    }
+
+    /// Returns the peer info in the trailers, if present.  Peer information
+    /// will not be available in trailers in any the following circumstances:
+    ///
+    /// 1. A ResponseHeaders was already present on the response stream.
+    ///
+    /// 2. The error was generated locally on the client before a connection was
+    ///    chosen for the RPC.
+    pub fn peer_info(&self) -> &Option<PeerInfo> {
+        &self.peer_info
+    }
+
     pub(crate) fn into_parts(self) -> (crate::Result<()>, MetadataMap) {
         (self.status, self.metadata)
     }

grpc/src/client/stream_util.rs

@@ -37,6 +37,7 @@ use crate::client::SendOptions;
 use crate::client::SendStream;
 use crate::client::Trailers;
 use crate::client::interceptor::Intercept;
+use crate::core::PeerInfo;
 use crate::core::RecvMessage;
 use crate::core::SendMessage;
 
@@ -181,12 +182,15 @@ impl SendStream for NopSendStream {
 
 pub(crate) struct FailingRecvStream {
     status: Option<StatusError>,
+    peer_info: Option<PeerInfo>,
 }
 
 impl RecvStream for FailingRecvStream {
     async fn recv(&mut self, msg: &mut dyn RecvMessage) -> ResponseStreamItem {
         match self.status.take() {
-            Some(status) => ResponseStreamItem::Trailers(Trailers::new(Err(status))),
+            Some(status) => ResponseStreamItem::Trailers(
+                Trailers::new(Err(status)).with_peer_info(self.peer_info.take()),
+            ),
             None => ResponseStreamItem::StreamClosed,
         }
     }
@@ -195,11 +199,13 @@ impl RecvStream for FailingRecvStream {
 impl FailingRecvStream {
     pub(crate) fn new_stream_pair(
         status: StatusError,
+        peer_info: Option<PeerInfo>,
     ) -> (Box<dyn DynSendStream>, Box<dyn DynRecvStream>) {
         (
             Box::new(NopSendStream),
             Box::new(Self {
                 status: Some(status),
+                peer_info,
             }),
         )
     }
@@ -240,11 +246,11 @@ mod test {
         let scenarios = [
             vec![ResponseStreamItem::StreamClosed],
             vec![
-                ResponseStreamItem::Headers(ResponseHeaders::default()),
+                ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
                 ResponseStreamItem::StreamClosed,
             ],
             vec![
-                ResponseStreamItem::Headers(ResponseHeaders::default()),
+                ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
                 ResponseStreamItem::Message,
                 ResponseStreamItem::StreamClosed,
             ],
@@ -268,13 +274,13 @@ mod test {
     async fn test_validator_headers_repeated() {
         let scenarios = [
             vec![
-                ResponseStreamItem::Headers(ResponseHeaders::default()),
-                ResponseStreamItem::Headers(ResponseHeaders::default()),
+                ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
+                ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
             ],
             vec![
-                ResponseStreamItem::Headers(ResponseHeaders::default()),
+                ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
                 ResponseStreamItem::Message,
-                ResponseStreamItem::Headers(ResponseHeaders::default()),
+                ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
             ],
         ];
 
@@ -296,7 +302,7 @@ mod test {
         let scenarios = [
             vec![ResponseStreamItem::Trailers(Trailers::new(Ok(())))],
             vec![
-                ResponseStreamItem::Headers(ResponseHeaders::default()),
+                ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
                 ResponseStreamItem::Trailers(Trailers::new(Ok(()))),
             ],
         ];
@@ -317,7 +323,7 @@ mod test {
     #[tokio::test]
     async fn test_validator_unary_multiple_messages() {
         let scenarios = [vec![
-            ResponseStreamItem::Headers(ResponseHeaders::default()),
+            ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
             ResponseStreamItem::Message,
             ResponseStreamItem::Message,
         ]];
@@ -338,7 +344,7 @@ mod test {
     #[tokio::test]
     async fn test_validator_successful_stream() {
         let scenarios = [vec![
-            ResponseStreamItem::Headers(ResponseHeaders::default()),
+            ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
             ResponseStreamItem::Message,
             ResponseStreamItem::Message,
             ResponseStreamItem::Message,
@@ -358,7 +364,7 @@ mod test {
     #[tokio::test]
     async fn test_validator_erroring_stream() {
         let scenarios = [vec![
-            ResponseStreamItem::Headers(ResponseHeaders::default()),
+            ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
             ResponseStreamItem::Message,
             ResponseStreamItem::Message,
             ResponseStreamItem::Message,
@@ -384,7 +390,7 @@ mod test {
     #[tokio::test]
     async fn test_validator_successful_unary() {
         let scenarios = [vec![
-            ResponseStreamItem::Headers(ResponseHeaders::default()),
+            ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
             ResponseStreamItem::Message,
             ResponseStreamItem::Trailers(Trailers::new(Ok(()))),
         ]];
@@ -406,14 +412,14 @@ mod test {
                 StatusError::new(StatusCodeError::Aborted, "some err"),
             )))],
             vec![
-                ResponseStreamItem::Headers(ResponseHeaders::default()),
+                ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
                 ResponseStreamItem::Trailers(Trailers::new(Err(StatusError::new(
                     StatusCodeError::Aborted,
                     "some err",
                 )))),
             ],
             vec![
-                ResponseStreamItem::Headers(ResponseHeaders::default()),
+                ResponseStreamItem::Headers(ResponseHeaders::new(crate::core::test_peer_info())),
                 ResponseStreamItem::Message,
                 ResponseStreamItem::Trailers(Trailers::new(Err(StatusError::new(
                     StatusCodeError::Aborted,

grpc/src/client/subchannel.rs

@@ -57,7 +57,7 @@ use crate::client::transport::SecurityOpts;
 use crate::client::transport::TransportOptions;
 use crate::client::transport::http_connect::HttpConnectHandshaker;
 use crate::core::Address;
-use crate::credentials::SecurityInfo;
+use crate::core::PeerInfo;
 use crate::credentials::call::CallDetails;
 use crate::credentials::call::ClientConnectionSecurityInfo as CallClientConnectionSecurityInfo;
 use crate::credentials::common::Authority;
@@ -86,7 +86,7 @@ impl Backoff for NopBackoff {
 
 struct ReadyState {
     service: Box<dyn DynInvoke>,
-    security_info: SecurityInfo,
+    peer_info: PeerInfo,
     authority: Authority,
 }
 
@@ -195,11 +195,13 @@ impl DynInvoke for InternalSubchannel {
         };
 
         let fail_with = |status| -> (Box<dyn DynSendStream>, Box<dyn DynRecvStream>) {
-            FailingRecvStream::new_stream_pair(status)
+            FailingRecvStream::new_stream_pair(status, Some(state.peer_info.clone()))
         };
 
         if let Some(call_creds) = call_creds {
-            if call_creds.minimum_channel_security_level() > state.security_info.security_level() {
+            if call_creds.minimum_channel_security_level()
+                > state.peer_info.security_info().security_level()
+            {
                 return fail_with(StatusError::new(
                     StatusCodeError::Unauthenticated,
                     "transport: cannot send secure credentials on an insecure connection",
@@ -209,9 +211,9 @@ impl DynInvoke for InternalSubchannel {
             let call_details = create_call_details(&state.authority, headers.method_name());
 
             let channel_sec_info = CallClientConnectionSecurityInfo::new(
-                state.security_info.security_protocol(),
-                state.security_info.security_level(),
-                state.security_info.attributes().clone(),
+                state.peer_info.security_info().security_protocol(),
+                state.peer_info.security_info().security_level(),
+                state.peer_info.security_info().attributes().clone(),
             );
 
             if let Err(s) = call_creds
@@ -376,10 +378,10 @@ fn begin_connecting_if_idle(data: Arc<Mutex<InternalSubchannelData>>) {
             }
             result = transport_builder.dyn_connect(&address, runtime, &security_opts, &transport_opts) => {
                     match result {
-                        Ok((service, security_info, disconnection_listener)) => {
+                        Ok((service, peer_info, disconnection_listener)) => {
                             move_to_ready(data, Arc::new(ReadyState{
                                 service,
-                                security_info,
+                                peer_info,
                                 authority: security_opts.authority}), disconnection_listener).await;
                         }
                         Err(e) => {

grpc/src/client/transport/mod.rs

@@ -30,8 +30,8 @@ use http::HeaderValue;
 use crate::client::DynInvoke;
 use crate::client::Invoke;
 use crate::core::Address;
+use crate::core::PeerInfo;
 use crate::credentials::ChannelCredentials;
-use crate::credentials::SecurityInfo;
 use crate::credentials::client::ClientHandshakeInfo;
 use crate::credentials::common::Authority;
 use crate::rt::GrpcRuntime;
@@ -98,7 +98,7 @@ pub(crate) trait Transport: Sync {
     ) -> Result<
         (
             Self::Service,
-            SecurityInfo,
+            PeerInfo,
             oneshot::Receiver<Result<(), String>>,
         ),
         String,
@@ -116,7 +116,7 @@ pub(crate) trait DynTransport: Send + Sync {
     ) -> Result<
         (
             Box<dyn DynInvoke>,
-            SecurityInfo,
+            PeerInfo,
             oneshot::Receiver<Result<(), String>>,
         ),
         String,
@@ -134,7 +134,7 @@ impl<T: Transport> DynTransport for T {
     ) -> Result<
         (
             Box<dyn DynInvoke>,
-            SecurityInfo,
+            PeerInfo,
             oneshot::Receiver<Result<(), String>>,
         ),
         String,