netty,okhttp,testing: always set TRANSPORT_ATTR_REMOTE_ADDR

netty/src/main/java/io/grpc/netty/ProtocolNegotiators.java

@@ -329,7 +329,7 @@ public Handler newHandler(GrpcHttp2ConnectionHandler handler) {
       HttpClientCodec httpClientCodec = new HttpClientCodec();
       final HttpClientUpgradeHandler upgrader =
           new HttpClientUpgradeHandler(httpClientCodec, upgradeCodec, 1000);
-      return new BufferingHttp2UpgradeHandler(upgrader);
+      return new BufferingHttp2UpgradeHandler(upgrader, handler);
     }
   }
 
@@ -662,8 +662,11 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
   private static class BufferUntilChannelActiveHandler extends AbstractBufferingHandler
       implements ProtocolNegotiator.Handler {
 
-    BufferUntilChannelActiveHandler(ChannelHandler... handlers) {
-      super(handlers);
+    private final GrpcHttp2ConnectionHandler handler;
+
+    BufferUntilChannelActiveHandler(GrpcHttp2ConnectionHandler handler) {
+      super(handler);
+      this.handler = handler;
     }
 
     @Override
@@ -679,6 +682,11 @@ public void handlerAdded(ChannelHandlerContext ctx) {
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
       writeBufferedAndRemove(ctx);
+      handler.handleProtocolNegotiationCompleted(
+          Attributes
+              .newBuilder()
+              .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, ctx.channel().remoteAddress())
+              .build());
       super.channelActive(ctx);
     }
   }
@@ -689,8 +697,11 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
   private static class BufferingHttp2UpgradeHandler extends AbstractBufferingHandler
       implements ProtocolNegotiator.Handler {
 
-    BufferingHttp2UpgradeHandler(ChannelHandler... handlers) {
-      super(handlers);
+    private final GrpcHttp2ConnectionHandler grpcHandler;
+
+    BufferingHttp2UpgradeHandler(ChannelHandler handler, GrpcHttp2ConnectionHandler grpcHandler) {
+      super(handler);
+      this.grpcHandler = grpcHandler;
     }
 
     @Override
@@ -712,6 +723,11 @@ public void channelActive(ChannelHandlerContext ctx) throws Exception {
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
       if (evt == HttpClientUpgradeHandler.UpgradeEvent.UPGRADE_SUCCESSFUL) {
         writeBufferedAndRemove(ctx);
+        grpcHandler.handleProtocolNegotiationCompleted(
+            Attributes
+                .newBuilder()
+                .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, ctx.channel().remoteAddress())
+                .build());
       } else if (evt == HttpClientUpgradeHandler.UpgradeEvent.UPGRADE_REJECTED) {
         fail(ctx, unavailableException("HTTP/2 upgrade rejected"));
       }

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java

@@ -123,7 +123,7 @@ public void setAuthority(String authority) {
 
   @Override
   public Attributes getAttributes() {
-    return Attributes.EMPTY;
+    return transportState().getAttributes();
   }
 
   class Sink implements AbstractClientStream.Sink {
@@ -318,6 +318,12 @@ public void transportDataReceived(okio.Buffer frame, boolean endOfStream) {
       super.transportDataReceived(new OkHttpReadableBuffer(frame), endOfStream);
     }
 
+    // Safe to call without lock
+    @SuppressWarnings("GuardedBy")
+    public Attributes getAttributes() {
+      return transport.getAttributes();
+    }
+
     @GuardedBy("lock")
     private void onEndOfStream() {
       if (!framer().isClosed()) {

okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java

@@ -31,6 +31,7 @@
 import com.squareup.okhttp.internal.http.StatusLine;
 import io.grpc.Attributes;
 import io.grpc.CallOptions;
+import io.grpc.Grpc;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.MethodDescriptor.MethodType;
@@ -149,6 +150,8 @@ private static Map<ErrorCode, Status> buildErrorCodeToStatusMap() {
   private final int maxMessageSize;
   private int connectionUnacknowledgedBytesRead;
   private ClientFrameHandler clientFrameHandler;
+  // Caution: Not synchronized, new value can only be safely read after the connection is complete.
+  private Attributes attributes = Attributes.EMPTY;
   /**
    * Indicates the transport is in go-away state: no new streams will be processed, but existing
    * streams may continue.
@@ -451,6 +454,7 @@ public void close() {}
         Variant variant = new Http2();
         BufferedSink sink;
         Socket sock;
+        Attributes attrs;
         try {
           if (proxy == null) {
             sock = new Socket(address.getAddress(), address.getPort());
@@ -467,6 +471,10 @@ sslSocketFactory, hostnameVerifier, sock, getOverridenHost(), getOverridenPort()
           sock.setTcpNoDelay(true);
           source = Okio.buffer(Okio.source(sock));
           sink = Okio.buffer(Okio.sink(sock));
+          attrs = Attributes
+              .newBuilder()
+              .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, sock.getRemoteSocketAddress())
+              .build();
         } catch (StatusException e) {
           startGoAway(0, ErrorCode.INTERNAL_ERROR, e.getStatus());
           return;
@@ -482,6 +490,11 @@ sslSocketFactory, hostnameVerifier, sock, getOverridenHost(), getOverridenPort()
         synchronized (lock) {
           socket = sock;
           maxConcurrentStreams = Integer.MAX_VALUE;
+
+          // TODO(zhangkun83): fill channel security attributes
+          // The return value of OkHttpTlsUpgrader.upgrade is an SSLSocket that has this info
+          attributes = attrs;
+
           startPendingStreams();
         }
 
@@ -675,8 +688,7 @@ public void shutdownNow(Status reason) {
 
   @Override
   public Attributes getAttributes() {
-    // TODO(zhangkun83): fill channel security attributes
-    return Attributes.EMPTY;
+    return attributes;
   }
 
   /**

testing/src/main/java/io/grpc/internal/testing/AbstractTransportTest.java

@@ -68,8 +68,6 @@
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.Arrays;
 import java.util.List;
@@ -1709,9 +1707,7 @@ public void socketStats_addresses() throws Exception {
         = serverTransportListener.takeStreamOrFail(TIMEOUT_MS, TimeUnit.MILLISECONDS);
     ServerStream serverStream = serverStreamCreation.stream;
 
-    // clients do not have TRANSPORT_ATTR_REMOTE_ADDR so use a hack for serverAddress
-    SocketAddress serverAddress
-        = new InetSocketAddress(InetAddress.getByName("127.0.0.1"), server.getPort());
+    SocketAddress serverAddress = clientStream.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
     SocketAddress clientAddress = serverStream.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
 
     SocketStats clientSocketStats = client.getStats().get();