gRPC support request/response size

instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/rpc/RpcAttributesGetter.java

@@ -24,4 +24,12 @@ public interface RpcAttributesGetter<REQUEST> {
 
   @Nullable
   String getMethod(REQUEST request);
+
+  default Long getRequestSize(REQUEST request) {
+    return null;
+  }
+
+  default Long getResponseSize(REQUEST request) {
+    return null;
+  }
 }

instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/rpc/RpcClientMetrics.java

@@ -11,6 +11,8 @@
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.metrics.DoubleHistogram;
 import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
+import io.opentelemetry.api.metrics.LongHistogram;
+import io.opentelemetry.api.metrics.LongHistogramBuilder;
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.ContextKey;
@@ -35,6 +37,8 @@ public final class RpcClientMetrics implements OperationListener {
   private static final Logger logger = Logger.getLogger(RpcClientMetrics.class.getName());
 
   private final DoubleHistogram clientDurationHistogram;
+  private final LongHistogram clientRequestSize;
+  private final LongHistogram clientResponseSize;
 
   private RpcClientMetrics(Meter meter) {
     DoubleHistogramBuilder durationBuilder =
@@ -44,6 +48,24 @@ private RpcClientMetrics(Meter meter) {
             .setUnit("ms");
     RpcMetricsAdvice.applyClientDurationAdvice(durationBuilder);
     clientDurationHistogram = durationBuilder.build();
+
+    LongHistogramBuilder requestSizeBuilder =
+        meter
+            .histogramBuilder("rpc.client.request.size")
+            .setUnit("By")
+            .setDescription("Measures the size of RPC request messages (uncompressed).")
+            .ofLongs();
+    RpcMetricsAdvice.applyClientRequestSizeAdvice(requestSizeBuilder);
+    clientRequestSize = requestSizeBuilder.build();
+
+    LongHistogramBuilder responseSizeBuilder =
+        meter
+            .histogramBuilder("rpc.client.response.size")
+            .setUnit("By")
+            .setDescription("Measures the size of RPC response messages (uncompressed).")
+            .ofLongs();
+    RpcMetricsAdvice.applyClientRequestSizeAdvice(responseSizeBuilder);
+    clientResponseSize = responseSizeBuilder.build();
   }
 
   /**
@@ -72,10 +94,21 @@ public void onEnd(Context context, Attributes endAttributes, long endNanos) {
           context);
       return;
     }
+    Attributes attributes = state.startAttributes().toBuilder().putAll(endAttributes).build();
     clientDurationHistogram.record(
-        (endNanos - state.startTimeNanos()) / NANOS_PER_MS,
-        state.startAttributes().toBuilder().putAll(endAttributes).build(),
-        context);
+        (endNanos - state.startTimeNanos()) / NANOS_PER_MS, attributes, context);
+
+    Long rpcClientRequestBodySize =
+        RpcMessageBodySizeUtil.getRpcRequestBodySize(endAttributes, state.startAttributes());
+    if (rpcClientRequestBodySize != null) {
+      clientRequestSize.record(rpcClientRequestBodySize, attributes, context);
+    }
+
+    Long rpcClientResponseBodySize =
+        RpcMessageBodySizeUtil.getRpcResponseBodySize(endAttributes, state.startAttributes());
+    if (rpcClientResponseBodySize != null) {
+      clientResponseSize.record(rpcClientResponseBodySize, attributes, context);
+    }
   }
 
   @AutoValue

instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/rpc/RpcCommonAttributesExtractor.java

@@ -20,6 +20,10 @@ abstract class RpcCommonAttributesExtractor<REQUEST, RESPONSE>
   static final AttributeKey<String> RPC_METHOD = AttributeKey.stringKey("rpc.method");
   static final AttributeKey<String> RPC_SERVICE = AttributeKey.stringKey("rpc.service");
   static final AttributeKey<String> RPC_SYSTEM = AttributeKey.stringKey("rpc.system");
+  static final AttributeKey<Long> RPC_REQUEST_BODY_SIZE =
+      AttributeKey.longKey("rpc.request.body.size");
+  static final AttributeKey<Long> RPC_RESPONSE_BODY_SIZE =
+      AttributeKey.longKey("rpc.response.body.size");
 
   private final RpcAttributesGetter<REQUEST> getter;
 
@@ -41,6 +45,24 @@ public final void onEnd(
       REQUEST request,
       @Nullable RESPONSE response,
       @Nullable Throwable error) {
-    // No response attributes
+    Long requestSize = getter.getRequestSize(request);
+    Long responseSize = getter.getResponseSize(request);
+    if (this instanceof RpcClientAttributesExtractor) {
+      if (requestSize != null) {
+        internalSet(attributes, RPC_REQUEST_BODY_SIZE, requestSize);
+      }
+      if (responseSize != null) {
+        internalSet(attributes, RPC_RESPONSE_BODY_SIZE, responseSize);
+      }
+    }
+
+    if (this instanceof RpcServerAttributesExtractor) {
+      if (requestSize != null) {
+        internalSet(attributes, RPC_REQUEST_BODY_SIZE, requestSize);
+      }
+      if (responseSize != null) {
+        internalSet(attributes, RPC_RESPONSE_BODY_SIZE, responseSize);
+      }
+    }
   }
 }

instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/rpc/RpcMessageBodySizeUtil.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.api.incubator.semconv.rpc;
+
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.common.Attributes;
+import javax.annotation.Nullable;
+
+final class RpcMessageBodySizeUtil {
+
+  @Nullable
+  static Long getRpcRequestBodySize(Attributes... attributesList) {
+    return getAttribute(RpcCommonAttributesExtractor.RPC_REQUEST_BODY_SIZE, attributesList);
+  }
+
+  @Nullable
+  static Long getRpcResponseBodySize(Attributes... attributesList) {
+    return getAttribute(RpcCommonAttributesExtractor.RPC_RESPONSE_BODY_SIZE, attributesList);
+  }
+
+  @Nullable
+  private static <T> T getAttribute(AttributeKey<T> key, Attributes... attributesList) {
+    for (Attributes attributes : attributesList) {
+      T value = attributes.get(key);
+      if (value != null) {
+        return value;
+      }
+    }
+    return null;
+  }
+
+  private RpcMessageBodySizeUtil() {}
+}

instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/rpc/RpcMetricsAdvice.java

@@ -5,36 +5,40 @@
 
 package io.opentelemetry.instrumentation.api.incubator.semconv.rpc;
 
+import static java.util.Arrays.asList;
+
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.incubator.metrics.ExtendedDoubleHistogramBuilder;
+import io.opentelemetry.api.incubator.metrics.ExtendedLongHistogramBuilder;
 import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
+import io.opentelemetry.api.metrics.LongHistogramBuilder;
 import io.opentelemetry.semconv.NetworkAttributes;
 import io.opentelemetry.semconv.ServerAttributes;
-import java.util.Arrays;
+import java.util.List;
 
 final class RpcMetricsAdvice {
 
   // copied from RpcIncubatingAttributes
   private static final AttributeKey<Long> RPC_GRPC_STATUS_CODE =
       AttributeKey.longKey("rpc.grpc.status_code");
+  private static final List<AttributeKey<?>> RPC_METRICS_ATTRIBUTE_KEYS =
+      asList(
+          RpcCommonAttributesExtractor.RPC_SYSTEM,
+          RpcCommonAttributesExtractor.RPC_SERVICE,
+          RpcCommonAttributesExtractor.RPC_METHOD,
+          RPC_GRPC_STATUS_CODE,
+          NetworkAttributes.NETWORK_TYPE,
+          NetworkAttributes.NETWORK_TRANSPORT,
+          ServerAttributes.SERVER_ADDRESS,
+          ServerAttributes.SERVER_PORT);
 
   static void applyClientDurationAdvice(DoubleHistogramBuilder builder) {
     if (!(builder instanceof ExtendedDoubleHistogramBuilder)) {
       return;
     }
     // the list of recommended metrics attributes is from
     // https://github.com/open-telemetry/semantic-conventions/blob/main/docs/rpc/rpc-metrics.md
-    ((ExtendedDoubleHistogramBuilder) builder)
-        .setAttributesAdvice(
-            Arrays.asList(
-                RpcCommonAttributesExtractor.RPC_SYSTEM,
-                RpcCommonAttributesExtractor.RPC_SERVICE,
-                RpcCommonAttributesExtractor.RPC_METHOD,
-                RPC_GRPC_STATUS_CODE,
-                NetworkAttributes.NETWORK_TYPE,
-                NetworkAttributes.NETWORK_TRANSPORT,
-                ServerAttributes.SERVER_ADDRESS,
-                ServerAttributes.SERVER_PORT));
+    ((ExtendedDoubleHistogramBuilder) builder).setAttributesAdvice(RPC_METRICS_ATTRIBUTE_KEYS);
   }
 
   static void applyServerDurationAdvice(DoubleHistogramBuilder builder) {
@@ -43,17 +47,25 @@ static void applyServerDurationAdvice(DoubleHistogramBuilder builder) {
     }
     // the list of recommended metrics attributes is from
     // https://github.com/open-telemetry/semantic-conventions/blob/main/docs/rpc/rpc-metrics.md
-    ((ExtendedDoubleHistogramBuilder) builder)
-        .setAttributesAdvice(
-            Arrays.asList(
-                RpcCommonAttributesExtractor.RPC_SYSTEM,
-                RpcCommonAttributesExtractor.RPC_SERVICE,
-                RpcCommonAttributesExtractor.RPC_METHOD,
-                RPC_GRPC_STATUS_CODE,
-                NetworkAttributes.NETWORK_TYPE,
-                NetworkAttributes.NETWORK_TRANSPORT,
-                ServerAttributes.SERVER_ADDRESS,
-                ServerAttributes.SERVER_PORT));
+    ((ExtendedDoubleHistogramBuilder) builder).setAttributesAdvice(RPC_METRICS_ATTRIBUTE_KEYS);
+  }
+
+  static void applyClientRequestSizeAdvice(LongHistogramBuilder builder) {
+    if (!(builder instanceof ExtendedLongHistogramBuilder)) {
+      return;
+    }
+    // the list of recommended metrics attributes is from
+    // https://github.com/open-telemetry/semantic-conventions/blob/main/docs/rpc/rpc-metrics.md
+    ((ExtendedLongHistogramBuilder) builder).setAttributesAdvice(RPC_METRICS_ATTRIBUTE_KEYS);
+  }
+
+  static void applyServerRequestSizeAdvice(LongHistogramBuilder builder) {
+    if (!(builder instanceof ExtendedLongHistogramBuilder)) {
+      return;
+    }
+    // the list of recommended metrics attributes is from
+    // https://github.com/open-telemetry/semantic-conventions/blob/main/docs/rpc/rpc-metrics.md
+    ((ExtendedLongHistogramBuilder) builder).setAttributesAdvice(RPC_METRICS_ATTRIBUTE_KEYS);
   }
 
   private RpcMetricsAdvice() {}

instrumentation-api-incubator/src/main/java/io/opentelemetry/instrumentation/api/incubator/semconv/rpc/RpcServerMetrics.java

@@ -11,6 +11,8 @@
 import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.metrics.DoubleHistogram;
 import io.opentelemetry.api.metrics.DoubleHistogramBuilder;
+import io.opentelemetry.api.metrics.LongHistogram;
+import io.opentelemetry.api.metrics.LongHistogramBuilder;
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.ContextKey;
@@ -35,6 +37,8 @@ public final class RpcServerMetrics implements OperationListener {
   private static final Logger logger = Logger.getLogger(RpcServerMetrics.class.getName());
 
   private final DoubleHistogram serverDurationHistogram;
+  private final LongHistogram serverRequestSize;
+  private final LongHistogram serverResponseSize;
 
   private RpcServerMetrics(Meter meter) {
     DoubleHistogramBuilder durationBuilder =
@@ -44,6 +48,24 @@ private RpcServerMetrics(Meter meter) {
             .setUnit("ms");
     RpcMetricsAdvice.applyServerDurationAdvice(durationBuilder);
     serverDurationHistogram = durationBuilder.build();
+
+    LongHistogramBuilder requestSizeBuilder =
+        meter
+            .histogramBuilder("rpc.server.request.size")
+            .setUnit("By")
+            .setDescription("Measures the size of RPC request messages (uncompressed).")
+            .ofLongs();
+    RpcMetricsAdvice.applyServerRequestSizeAdvice(requestSizeBuilder);
+    serverRequestSize = requestSizeBuilder.build();
+
+    LongHistogramBuilder responseSizeBuilder =
+        meter
+            .histogramBuilder("rpc.server.response.size")
+            .setUnit("By")
+            .setDescription("Measures the size of RPC response messages (uncompressed).")
+            .ofLongs();
+    RpcMetricsAdvice.applyServerRequestSizeAdvice(responseSizeBuilder);
+    serverResponseSize = responseSizeBuilder.build();
   }
 
   /**
@@ -72,10 +94,21 @@ public void onEnd(Context context, Attributes endAttributes, long endNanos) {
           context);
       return;
     }
+    Attributes attributes = state.startAttributes().toBuilder().putAll(endAttributes).build();
     serverDurationHistogram.record(
-        (endNanos - state.startTimeNanos()) / NANOS_PER_MS,
-        state.startAttributes().toBuilder().putAll(endAttributes).build(),
-        context);
+        (endNanos - state.startTimeNanos()) / NANOS_PER_MS, attributes, context);
+
+    Long rpcServerRequestBodySize =
+        RpcMessageBodySizeUtil.getRpcRequestBodySize(endAttributes, state.startAttributes());
+    if (rpcServerRequestBodySize != null) {
+      serverRequestSize.record(rpcServerRequestBodySize, attributes, context);
+    }
+
+    Long rpcServerResponseBodySize =
+        RpcMessageBodySizeUtil.getRpcResponseBodySize(endAttributes, state.startAttributes());
+    if (rpcServerResponseBodySize != null) {
+      serverResponseSize.record(rpcServerResponseBodySize, attributes, context);
+    }
   }
 
   @AutoValue

instrumentation-api-incubator/src/test/java/io/opentelemetry/instrumentation/api/incubator/semconv/rpc/RpcClientMetricsTest.java

@@ -46,6 +46,8 @@ void collectsMetrics() {
             .put(ServerAttributes.SERVER_PORT, 8080)
             .put(NetworkAttributes.NETWORK_TRANSPORT, "tcp")
             .put(NetworkAttributes.NETWORK_TYPE, "ipv4")
+            .put(RpcCommonAttributesExtractor.RPC_REQUEST_BODY_SIZE, 10)
+            .put(RpcCommonAttributesExtractor.RPC_RESPONSE_BODY_SIZE, 20)
             .build();
 
     Attributes responseAttributes2 =
@@ -76,6 +78,62 @@ void collectsMetrics() {
 
     assertThat(metricReader.collectAllMetrics())
         .satisfiesExactlyInAnyOrder(
+            metric ->
+                assertThat(metric)
+                    .hasName("rpc.client.response.size")
+                    .hasUnit("By")
+                    .hasDescription("Measures the size of RPC response messages (uncompressed).")
+                    .hasHistogramSatisfying(
+                        histogram ->
+                            histogram.hasPointsSatisfying(
+                                point ->
+                                    point
+                                        .hasSum(20 /* bytes */)
+                                        .hasAttributesSatisfying(
+                                            equalTo(RpcIncubatingAttributes.RPC_SYSTEM, "grpc"),
+                                            equalTo(
+                                                RpcIncubatingAttributes.RPC_SERVICE,
+                                                "myservice.EchoService"),
+                                            equalTo(
+                                                RpcIncubatingAttributes.RPC_METHOD,
+                                                "exampleMethod"),
+                                            equalTo(ServerAttributes.SERVER_ADDRESS, "example.com"),
+                                            equalTo(ServerAttributes.SERVER_PORT, 8080),
+                                            equalTo(NetworkAttributes.NETWORK_TRANSPORT, "tcp"),
+                                            equalTo(NetworkAttributes.NETWORK_TYPE, "ipv4"))
+                                        .hasExemplarsSatisfying(
+                                            exemplar ->
+                                                exemplar
+                                                    .hasTraceId("ff01020304050600ff0a0b0c0d0e0f00")
+                                                    .hasSpanId("090a0b0c0d0e0f00")))),
+            metric ->
+                assertThat(metric)
+                    .hasName("rpc.client.request.size")
+                    .hasUnit("By")
+                    .hasDescription("Measures the size of RPC request messages (uncompressed).")
+                    .hasHistogramSatisfying(
+                        histogram ->
+                            histogram.hasPointsSatisfying(
+                                point ->
+                                    point
+                                        .hasSum(10 /* bytes */)
+                                        .hasAttributesSatisfying(
+                                            equalTo(RpcIncubatingAttributes.RPC_SYSTEM, "grpc"),
+                                            equalTo(
+                                                RpcIncubatingAttributes.RPC_SERVICE,
+                                                "myservice.EchoService"),
+                                            equalTo(
+                                                RpcIncubatingAttributes.RPC_METHOD,
+                                                "exampleMethod"),
+                                            equalTo(ServerAttributes.SERVER_ADDRESS, "example.com"),
+                                            equalTo(ServerAttributes.SERVER_PORT, 8080),
+                                            equalTo(NetworkAttributes.NETWORK_TRANSPORT, "tcp"),
+                                            equalTo(NetworkAttributes.NETWORK_TYPE, "ipv4"))
+                                        .hasExemplarsSatisfying(
+                                            exemplar ->
+                                                exemplar
+                                                    .hasTraceId("ff01020304050600ff0a0b0c0d0e0f00")
+                                                    .hasSpanId("090a0b0c0d0e0f00")))),
             metric ->
                 assertThat(metric)
                     .hasName("rpc.client.duration")