core: record stats for GET requests (#2914)

Also define the expected call sites of StatsTraceContext.

core/src/main/java/io/grpc/internal/AbstractClientStream2.java

@@ -349,6 +349,18 @@ public abstract class AbstractClientStream2 extends AbstractStream2
       } catch (java.io.IOException ex) {
         throw new RuntimeException(ex);
       }
+      statsTraceCtx.outboundMessage();
+      statsTraceCtx.outboundUncompressedSize(payload.length);
+      // NB(zhangkun83): this is not accurate, because the underlying transport will probably encode
+      // it using e.g., base64.  However, we are not supposed to know such detail here.
+      //
+      // We don't want to move this line to where the encoding happens either, because we'd better
+      // contain the message stats reporting in Framer as suggested in StatsTraceContext.
+      // Scattering the reporting sites increases the risk of mis-counting or double-counting.
+      //
+      // Because the payload is usually very small, people shouldn't care about the size difference
+      // caused by encoding.
+      statsTraceCtx.outboundWireSize(payload.length);
     }
 
     @Override

core/src/main/java/io/grpc/internal/StatsTraceContext.java

@@ -104,6 +104,8 @@ public final class StatsTraceContext {
 
   /**
    * See {@link ClientStreamTracer#headersSent}.  For client-side only.
+   *
+   * <p>Transport-specific, thus should be called by transport implementations.
    */
   public void clientHeadersSent() {
     for (StreamTracer tracer : tracers) {
@@ -113,6 +115,8 @@ public final class StatsTraceContext {
 
   /**
    * See {@link ServerStreamTracer#filterContext}.  For server-side only.
+   *
+   * <p>Called from {@link io.grpc.internal.ServerImpl}.
    */
   public <ReqT, RespT> Context serverFilterContext(Context context) {
     Context ctx = checkNotNull(context, "context");
@@ -126,6 +130,8 @@ public final class StatsTraceContext {
   /**
    * See {@link StreamTracer#streamClosed}. This may be called multiple times, and only the first
    * value will be taken.
+   *
+   * <p>Called from abstract stream implementations.
    */
   public void streamClosed(Status status) {
     if (closed.compareAndSet(false, true)) {
@@ -137,6 +143,8 @@ public final class StatsTraceContext {
 
   /**
    * See {@link StreamTracer#outboundMessage}.
+   *
+   * <p>Called from {@link io.grpc.internal.Framer}.
    */
   public void outboundMessage() {
     for (StreamTracer tracer : tracers) {
@@ -146,6 +154,8 @@ public final class StatsTraceContext {
 
   /**
    * See {@link StreamTracer#inboundMessage}.
+   *
+   * <p>Called from {@link io.grpc.internal.MessageDeframer}.
    */
   public void inboundMessage() {
     for (StreamTracer tracer : tracers) {
@@ -155,6 +165,8 @@ public final class StatsTraceContext {
 
   /**
    * See {@link StreamTracer#outboundUncompressedSize}.
+   *
+   * <p>Called from {@link io.grpc.internal.Framer}.
    */
   public void outboundUncompressedSize(long bytes) {
     for (StreamTracer tracer : tracers) {
@@ -164,6 +176,8 @@ public final class StatsTraceContext {
 
   /**
    * See {@link StreamTracer#outboundWireSize}.
+   *
+   * <p>Called from {@link io.grpc.internal.Framer}.
    */
   public void outboundWireSize(long bytes) {
     for (StreamTracer tracer : tracers) {
@@ -173,6 +187,8 @@ public final class StatsTraceContext {
 
   /**
    * See {@link StreamTracer#inboundUncompressedSize}.
+   *
+   * <p>Called from {@link io.grpc.internal.MessageDeframer}.
    */
   public void inboundUncompressedSize(long bytes) {
     for (StreamTracer tracer : tracers) {
@@ -182,6 +198,8 @@ public final class StatsTraceContext {
 
   /**
    * See {@link StreamTracer#inboundWireSize}.
+   *
+   * <p>Called from {@link io.grpc.internal.MessageDeframer}.
    */
   public void inboundWireSize(long bytes) {
     for (StreamTracer tracer : tracers) {

core/src/test/java/io/grpc/internal/AbstractClientStream2Test.java

@@ -37,13 +37,17 @@ import static org.junit.Assert.fail;
 import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 
 import io.grpc.Attributes;
+import io.grpc.ClientStreamTracer;
 import io.grpc.Codec;
 import io.grpc.Metadata;
 import io.grpc.Status;
 import io.grpc.Status.Code;
+import io.grpc.StreamTracer;
 import io.grpc.internal.AbstractClientStream2.TransportState;
 import io.grpc.internal.MessageFramerTest.ByteWritableBuffer;
 import java.io.ByteArrayInputStream;
@@ -223,6 +227,15 @@ public class AbstractClientStream2Test {
   @Test
   public void getRequest() {
     AbstractClientStream2.Sink sink = mock(AbstractClientStream2.Sink.class);
+    final ClientStreamTracer tracer = spy(new ClientStreamTracer() {});
+    ClientStreamTracer.Factory tracerFactory =
+        new ClientStreamTracer.Factory() {
+          @Override
+          public ClientStreamTracer newClientStreamTracer(Metadata headers) {
+            return tracer;
+          }
+        };
+    StatsTraceContext statsTraceCtx = new StatsTraceContext(new StreamTracer[] {tracer});
     AbstractClientStream2 stream = new BaseAbstractClientStream(allocator,
         new BaseTransportState(statsTraceCtx), sink, statsTraceCtx, true);
     stream.start(mockListener);
@@ -237,6 +250,10 @@ public class AbstractClientStream2Test {
     // GET requests don't have BODY.
     verify(sink, never())
         .writeFrame(any(WritableBuffer.class), any(Boolean.class), any(Boolean.class));
+    verify(tracer).outboundMessage();
+    verify(tracer).outboundWireSize(1);
+    verify(tracer).outboundUncompressedSize(1);
+    verifyNoMoreInteractions(tracer);
   }
 
   /**