fix(ccs): Reports gRPC status code in CCS

# Conflicts:
#	dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricKey.java

Author: bric3

dd-trace-core/src/main/java/datadog/trace/common/metrics/ConflatingMetricsAggregator.java

@@ -1,6 +1,7 @@
 package datadog.trace.common.metrics;
 
 import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V06_METRICS_ENDPOINT;
+import static datadog.trace.api.DDSpanTypes.RPC;
 import static datadog.trace.api.DDTags.BASE_SERVICE;
 import static datadog.trace.api.Functions.UTF8_ENCODE;
 import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_ENDPOINT;
@@ -79,6 +80,7 @@ public final class ConflatingMetricsAggregator implements MetricsAggregator, Eve
                   DDCaches.newFixedSizeCache(512),
                   value -> UTF8BytesString.create(key + ":" + value));
   private static final CharSequence SYNTHETICS_ORIGIN = "synthetics";
+  private static final String GRPC_STATUS_TAG = "grpc.status.code";
 
   private static final Set<String> ELIGIBLE_SPAN_KINDS_FOR_METRICS =
       unmodifiableSet(
@@ -326,21 +328,28 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel, CharSequence spanK
       httpEndpoint = httpEndpointObj != null ? httpEndpointObj.toString() : null;
     }
 
+    CharSequence spanType = span.getType();
+    String grpcStatusCode = null;
+    if (spanType != null && RPC.contentEquals(spanType)) {
+      Object grpcStatusObj = span.unsafeGetTag(GRPC_STATUS_TAG);
+      grpcStatusCode = grpcStatusObj != null ? grpcStatusObj.toString() : null;
+    }
     MetricKey newKey =
         new MetricKey(
             span.getResourceName(),
             SERVICE_NAMES.computeIfAbsent(span.getServiceName(), UTF8_ENCODE),
             span.getOperationName(),
             span.getServiceNameSource(),
-            span.getType(),
+            spanType,
             span.getHttpStatusCode(),
             isSynthetic(span),
             span.getParentId() == 0,
             SPAN_KINDS.computeIfAbsent(
                 spanKind, UTF8BytesString::create), // save repeated utf8 conversions
             getPeerTags(span, spanKind.toString()),
             httpMethod,
-            httpEndpoint);
+            httpEndpoint,
+            grpcStatusCode);
     MetricKey key = keys.putIfAbsent(newKey, newKey);
     if (null == key) {
       key = newKey;

dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricKey.java

@@ -22,6 +22,8 @@ public final class MetricKey {
   static final DDCache<String, UTF8BytesString> HTTP_METHOD_CACHE = DDCaches.newFixedSizeCache(8);
   static final DDCache<String, UTF8BytesString> HTTP_ENDPOINT_CACHE =
       DDCaches.newFixedSizeCache(32);
+  static final DDCache<String, UTF8BytesString> GRPC_STATUS_CODE_CACHE =
+      DDCaches.newFixedSizeCache(32);
 
   private final UTF8BytesString resource;
   private final UTF8BytesString service;
@@ -36,6 +38,7 @@ public final class MetricKey {
   private final List<UTF8BytesString> peerTags;
   private final UTF8BytesString httpMethod;
   private final UTF8BytesString httpEndpoint;
+  private final UTF8BytesString grpcStatusCode;
 
   public MetricKey(
       CharSequence resource,
@@ -49,7 +52,8 @@ public MetricKey(
       CharSequence spanKind,
       List<UTF8BytesString> peerTags,
       CharSequence httpMethod,
-      CharSequence httpEndpoint) {
+      CharSequence httpEndpoint,
+      CharSequence grpcStatusCode) {
     this.resource = null == resource ? EMPTY : utf8(RESOURCE_CACHE, resource);
     this.service = null == service ? EMPTY : utf8(SERVICE_CACHE, service);
     this.serviceSource = null == serviceSource ? null : utf8(SERVICE_SOURCE_CACHE, serviceSource);
@@ -62,6 +66,8 @@ public MetricKey(
     this.peerTags = peerTags == null ? Collections.emptyList() : peerTags;
     this.httpMethod = httpMethod == null ? null : utf8(HTTP_METHOD_CACHE, httpMethod);
     this.httpEndpoint = httpEndpoint == null ? null : utf8(HTTP_ENDPOINT_CACHE, httpEndpoint);
+    this.grpcStatusCode =
+        grpcStatusCode == null ? null : utf8(GRPC_STATUS_CODE_CACHE, grpcStatusCode);
 
     int tmpHash = 0;
     tmpHash = HashingUtils.addToHash(tmpHash, this.isTraceRoot);
@@ -76,6 +82,7 @@ public MetricKey(
     tmpHash = HashingUtils.addToHash(tmpHash, this.serviceSource);
     tmpHash = HashingUtils.addToHash(tmpHash, this.httpEndpoint);
     tmpHash = HashingUtils.addToHash(tmpHash, this.httpMethod);
+    tmpHash = HashingUtils.addToHash(tmpHash, this.grpcStatusCode);
     this.hash = tmpHash;
   }
 
@@ -135,6 +142,10 @@ public UTF8BytesString getHttpEndpoint() {
     return httpEndpoint;
   }
 
+  public UTF8BytesString getGrpcStatusCode() {
+    return grpcStatusCode;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -154,7 +165,8 @@ public boolean equals(Object o) {
           && peerTags.equals(metricKey.peerTags)
           && Objects.equals(serviceSource, metricKey.serviceSource)
           && Objects.equals(httpMethod, metricKey.httpMethod)
-          && Objects.equals(httpEndpoint, metricKey.httpEndpoint);
+          && Objects.equals(httpEndpoint, metricKey.httpEndpoint)
+          && Objects.equals(grpcStatusCode, metricKey.grpcStatusCode);
     }
     return false;
   }

dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java

@@ -37,6 +37,7 @@ public final class SerializingMetricWriter implements MetricWriter {
   private static final byte[] PEER_TAGS = "PeerTags".getBytes(ISO_8859_1);
   private static final byte[] HTTP_METHOD = "HTTPMethod".getBytes(ISO_8859_1);
   private static final byte[] HTTP_ENDPOINT = "HTTPEndpoint".getBytes(ISO_8859_1);
+  private static final byte[] GRPC_STATUS_CODE = "GRPCStatusCode".getBytes(ISO_8859_1);
   private static final byte[] SERVICE_SOURCE = "srv_src".getBytes(ISO_8859_1);
 
   // Constant declared here for compile-time folding
@@ -111,8 +112,13 @@ public void add(MetricKey key, AggregateMetric aggregate) {
     final boolean hasHttpMethod = key.getHttpMethod() != null;
     final boolean hasHttpEndpoint = key.getHttpEndpoint() != null;
     final boolean hasServiceSource = key.getServiceSource() != null;
+    final boolean hasGrpcStatusCode = key.getGrpcStatusCode() != null;
     final int mapSize =
-        15 + (hasServiceSource ? 1 : 0) + (hasHttpMethod ? 1 : 0) + (hasHttpEndpoint ? 1 : 0);
+        15
+            + (hasServiceSource ? 1 : 0)
+            + (hasHttpMethod ? 1 : 0)
+            + (hasHttpEndpoint ? 1 : 0)
+            + (hasGrpcStatusCode ? 1 : 0);
 
     writer.startMap(mapSize);
 
@@ -164,6 +170,12 @@ public void add(MetricKey key, AggregateMetric aggregate) {
       writer.writeUTF8(key.getHttpEndpoint());
     }
 
+    // Only include GRPCStatusCode if present (rpc-type spans)
+    if (hasGrpcStatusCode) {
+      writer.writeUTF8(GRPC_STATUS_CODE);
+      writer.writeUTF8(key.getGrpcStatusCode());
+    }
+
     writer.writeUTF8(HITS);
     writer.writeInt(aggregate.getHitCount());
 

dd-trace-core/src/test/groovy/datadog/trace/common/metrics/AggregateMetricTest.groovy

@@ -65,7 +65,7 @@ class AggregateMetricTest extends DDSpecification {
     given:
     AggregateMetric aggregate = new AggregateMetric().recordDurations(3, new AtomicLongArray(0L, 0L, 0L | ERROR_TAG | TOP_LEVEL_TAG))
 
-    Batch batch = new Batch().reset(new MetricKey("foo", "bar", "qux", null, "type", 0, false, true, "corge", [UTF8BytesString.create("grault:quux")], null, null))
+    Batch batch = new Batch().reset(new MetricKey("foo", "bar", "qux", null, "type", 0, false, true, "corge", [UTF8BytesString.create("grault:quux")], null, null, null))
     batch.add(0L, 10)
     batch.add(0L, 10)
     batch.add(0L, 10)
@@ -140,7 +140,7 @@ class AggregateMetricTest extends DDSpecification {
   def "consistent under concurrent attempts to read and write"() {
     given:
     AggregateMetric aggregate = new AggregateMetric()
-    MetricKey key = new MetricKey("foo", "bar", "qux", null, "type", 0, false, true, "corge", [UTF8BytesString.create("grault:quux")], null, null)
+    MetricKey key = new MetricKey("foo", "bar", "qux", null, "type", 0, false, true, "corge", [UTF8BytesString.create("grault:quux")], null, null, null)
     BlockingDeque<Batch> queue = new LinkedBlockingDeque<>(1000)
     ExecutorService reader = Executors.newSingleThreadExecutor()
     int writerCount = 10