netty,okhttp,testing: always set TRANSPORT_ATTR_REMOTE_ADDR (#4217)

Always set the remote address, no reason why this should be a TLS-only
feature. This is needed for channelz, and is especially useful in unit
tests where we are using plaintext.

This PR adds the attr for plaintext.

2 files changed, 17 insertions, 3 deletions

diff --git a/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java b/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java
index 7a5ee95da..0ea8a89f0 100644
--- a/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java
+++ b/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java
@@ -58,6 +58,7 @@ class OkHttpClientStream extends AbstractClientStream {
   private volatile int id = ABSENT_ID;
   private final TransportState state;
   private final Sink sink = new Sink();
+  private final Attributes attributes;
 
   private boolean useGet = false;
 
@@ -83,6 +84,10 @@ class OkHttpClientStream extends AbstractClientStream {
     this.method = method;
     this.authority = authority;
     this.userAgent = userAgent;
+    // OkHttpClientStream is only created after the transport has finished connecting,
+    // so it is safe to read the transport attributes.
+    // We make a copy here for convenience, even though we can ask the transport.
+    this.attributes = transport.getAttributes();
     this.state = new TransportState(maxMessageSize, statsTraceCtx, lock, frameWriter, outboundFlow,
         transport);
   }
@@ -123,7 +128,7 @@ class OkHttpClientStream extends AbstractClientStream {
 
   @Override
   public Attributes getAttributes() {
-    return Attributes.EMPTY;
+    return attributes;
   }
 
   class Sink implements AbstractClientStream.Sink {
diff --git a/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java b/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java
index a1fc9fa1c..f710399c8 100644
--- a/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java
+++ b/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java
@@ -31,6 +31,7 @@ import com.squareup.okhttp.Request;
 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 @@ class OkHttpClientTransport implements ConnectionClientTransport {
   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.
@@ -467,6 +470,12 @@ class OkHttpClientTransport implements ConnectionClientTransport {
           sock.setTcpNoDelay(true);
           source = Okio.buffer(Okio.source(sock));
           sink = Okio.buffer(Okio.sink(sock));
+          // TODO(zhangkun83): fill channel security attributes
+          // The return value of OkHttpTlsUpgrader.upgrade is an SSLSocket that has this info
+          attributes = Attributes
+              .newBuilder()
+              .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, sock.getRemoteSocketAddress())
+              .build();
         } catch (StatusException e) {
           startGoAway(0, ErrorCode.INTERNAL_ERROR, e.getStatus());
           return;
@@ -482,6 +491,7 @@ class OkHttpClientTransport implements ConnectionClientTransport {
         synchronized (lock) {
           socket = sock;
           maxConcurrentStreams = Integer.MAX_VALUE;
+
           startPendingStreams();
         }
 
@@ -675,8 +685,7 @@ class OkHttpClientTransport implements ConnectionClientTransport {
 
   @Override
   public Attributes getAttributes() {
-    // TODO(zhangkun83): fill channel security attributes
-    return Attributes.EMPTY;
+    return attributes;
   }
 
   /**