chore(css): Use a map to create the metric key

Also delays the creation of the Utf8ByteString at serialization time.
Note that `writer.writeUTF8` emits a header corresponding to the length
of the value being written, calling this method again will count as an
entry in the array.
A possible idea was to use a cache to store the computation of the
peerTags, however `Utf8ByteString` is not concatenable / appendable,
which is necessary to have the proper encoding. Creating a
new "temporary" `Utf8ByteString`, was replaced by a direct call to
`String: :getBytes`.

Author: bric3

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

@@ -24,9 +24,8 @@
 import datadog.trace.core.CoreSpan;
 import datadog.trace.core.DDTraceCoreInfo;
 import datadog.trace.util.AgentTaskScheduler;
-import datadog.trace.util.TraceUtils;
-import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
@@ -254,7 +253,8 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel) {
             isSynthetic(span),
             span.isTopLevel(),
             span.getTag(SPAN_KIND, ""),
-            getPeerTags(span));
+            getPeerTags(span)
+        );
     boolean isNewKey = false;
     MetricKey key = keys.putIfAbsent(newKey, newKey);
     if (null == key) {
@@ -269,7 +269,7 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel) {
       // returning false means that either the batch can't take any
       // more data, or it has already been consumed
       if (batch.add(tag, durationNanos)) {
-        // added to a pending batch prior to consumption
+        // added to a pending batch prior to consumption,
         // so skip publishing to the queue (we also know
         // the key isn't rare enough to override the sampler)
         return false;
@@ -289,12 +289,12 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel) {
     return isNewKey || span.getError() > 0;
   }
 
-  private List<CharSequence> getPeerTags(CoreSpan<?> span) {
-    List<CharSequence> peerTags = new ArrayList<>();
+  private Map<String, String> getPeerTags(CoreSpan<?> span) {
+    Map <String, String> peerTags = new HashMap<>();
     for (String peerTag : features.peerTags()) {
       Object value = span.getTag(peerTag);
       if (value != null) {
-        peerTags.add(peerTag + ":" + TraceUtils.normalizeTag(value.toString()));
+        peerTags.put(peerTag, value.toString());
       }
     }
     return peerTags;

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

@@ -3,8 +3,8 @@
 import static datadog.trace.bootstrap.instrumentation.api.UTF8BytesString.EMPTY;
 
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
-import java.util.Arrays;
-import java.util.List;
+import java.util.Collections;
+import java.util.Map;
 
 /** The aggregation key for tracked metrics. */
 public final class MetricKey {
@@ -17,7 +17,7 @@ public final class MetricKey {
   private final int hash;
   private final boolean isTraceRoot;
   private final UTF8BytesString spanKind;
-  private final UTF8BytesString[] peerTags;
+  private final Map<String, String> peerTags;
 
   public MetricKey(
       CharSequence resource,
@@ -28,7 +28,8 @@ public MetricKey(
       boolean synthetics,
       boolean isTraceRoot,
       CharSequence spanKind,
-      List<CharSequence> peerTags) {
+      Map<String, String> peerTags
+  ) {
     this.resource = null == resource ? EMPTY : UTF8BytesString.create(resource);
     this.service = null == service ? EMPTY : UTF8BytesString.create(service);
     this.operationName = null == operationName ? EMPTY : UTF8BytesString.create(operationName);
@@ -37,21 +38,19 @@ public MetricKey(
     this.synthetics = synthetics;
     this.isTraceRoot = isTraceRoot;
     this.spanKind = null == spanKind ? EMPTY : UTF8BytesString.create(spanKind);
-    this.peerTags = new UTF8BytesString[peerTags.size()];
-    for (int i = 0; i < peerTags.size(); i++) {
-      this.peerTags[i] = UTF8BytesString.create(peerTags.get(i));
-    }
+    this.peerTags = peerTags ==  null ? Collections.emptyMap() : peerTags;
 
     // Unrolled polynomial hashcode to avoid varargs allocation
     // and eliminate data dependency between iterations as in Arrays.hashCode.
     // Coefficient constants are powers of 31, with integer overflow (hence negative numbers).
     // See
     // https://richardstartin.github.io/posts/collecting-rocks-and-benchmarks
     // https://richardstartin.github.io/posts/still-true-in-java-9-handwritten-hash-codes-are-faster
+
     this.hash =
         -196513505 * Boolean.hashCode(this.isTraceRoot)
             + -1807454463 * this.spanKind.hashCode()
-            + 887_503_681 * Arrays.hashCode(this.peerTags)
+            + 887_503_681 * this.peerTags.hashCode() // possibly unroll here has well.
             + 28_629_151 * this.resource.hashCode()
             + 923_521 * this.service.hashCode()
             + 29791 * this.operationName.hashCode()
@@ -92,7 +91,7 @@ public UTF8BytesString getSpanKind() {
     return spanKind;
   }
 
-  public UTF8BytesString[] getPeerTags() {
+  public Map<String, String> getPeerTags() {
     return peerTags;
   }
 
@@ -112,7 +111,7 @@ public boolean equals(Object o) {
           && type.equals(metricKey.type)
           && isTraceRoot == metricKey.isTraceRoot
           && spanKind.equals(metricKey.spanKind)
-          && Arrays.equals(peerTags, metricKey.peerTags);
+          && peerTags.equals(metricKey.peerTags);
     }
     return false;
   }

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

@@ -1,13 +1,17 @@
 package datadog.trace.common.metrics;
 
 import static java.nio.charset.StandardCharsets.ISO_8859_1;
+import static java.nio.charset.StandardCharsets.UTF_8;
 
 import datadog.communication.serialization.GrowableBuffer;
 import datadog.communication.serialization.WritableFormatter;
 import datadog.communication.serialization.msgpack.MsgPackWriter;
 import datadog.trace.api.ProcessTags;
 import datadog.trace.api.WellKnownTags;
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
+import datadog.trace.util.TraceUtils;
+
+import java.util.Map;
 
 public final class SerializingMetricWriter implements MetricWriter {
 
@@ -34,6 +38,7 @@ public final class SerializingMetricWriter implements MetricWriter {
   private static final byte[] IS_TRACE_ROOT = "IsTraceRoot".getBytes(ISO_8859_1);
   private static final byte[] SPAN_KIND = "SpanKind".getBytes(ISO_8859_1);
   private static final byte[] PEER_TAGS = "PeerTags".getBytes(ISO_8859_1);
+  private static final byte[] PEER_TAG_SEPARATOR = ":".getBytes(ISO_8859_1);
 
   private final WellKnownTags wellKnownTags;
   private final WritableFormatter writer;
@@ -123,10 +128,17 @@ public void add(MetricKey key, AggregateMetric aggregate) {
     writer.writeUTF8(key.getSpanKind());
 
     writer.writeUTF8(PEER_TAGS);
-    UTF8BytesString[] peerTags = key.getPeerTags();
-    writer.startArray(peerTags.length);
-    for (UTF8BytesString peerTag : peerTags) {
-      writer.writeUTF8(peerTag);
+    Map<String, String> peerTags = key.getPeerTags();
+    writer.startArray(peerTags.size());
+
+    StringBuilder peerTagBuilder = new StringBuilder();
+    for (Map.Entry<String, String> peerTag : peerTags.entrySet()) {
+      peerTagBuilder.setLength(0);
+      String toWrite = peerTagBuilder
+          .append(peerTag.getKey())
+          .append(':')
+          .append(TraceUtils.normalizeTag(peerTag.getValue())).toString();
+      writer.writeUTF8(toWrite.getBytes(UTF_8));
     }
 
     writer.writeUTF8(HITS);

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

@@ -51,7 +51,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", "type", 0, false, true, "corge", ["grault"]))
+    Batch batch = new Batch().reset(new MetricKey("foo", "bar", "qux", "type", 0, false, true, "corge", ["grault":"quux"]))
     batch.add(0L, 10)
     batch.add(0L, 10)
     batch.add(0L, 10)
@@ -126,7 +126,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", "type", 0, false, true, "corge", ["grault"])
+    MetricKey key = new MetricKey("foo", "bar", "qux", "type", 0, false, true, "corge", ["grault":"quux"])
     BlockingDeque<Batch> queue = new LinkedBlockingDeque<>(1000)
     ExecutorService reader = Executors.newSingleThreadExecutor()
     int writerCount = 10

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

@@ -125,7 +125,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       false,
       true,
       "baz",
-      []
+      [:]
       ), _) >> { MetricKey key, AggregateMetric value ->
         value.getHitCount() == 1 && value.getTopLevelCount() == 1 && value.getDuration() == 100
       }
@@ -168,7 +168,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       false,
       false,
       kind,
-      []
+      [:]
       ), { AggregateMetric aggregateMetric ->
         aggregateMetric.getHitCount() == 1 && aggregateMetric.getTopLevelCount() == 0 && aggregateMetric.getDuration() == 100
       })
@@ -221,7 +221,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       false,
       false,
       "grault",
-      ["country:france"]
+      ["country":"france"]
       ), { AggregateMetric aggregateMetric ->
         aggregateMetric.getHitCount() == 1 && aggregateMetric.getTopLevelCount() == 0 && aggregateMetric.getDuration() == 100
       })
@@ -235,7 +235,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       false,
       false,
       "grault",
-      ["country:france", "georegion:europe"]
+      ["country":"france", "georegion":"europe"]
       ), { AggregateMetric aggregateMetric ->
         aggregateMetric.getHitCount() == 1 && aggregateMetric.getTopLevelCount() == 0 && aggregateMetric.getDuration() == 100
       })
@@ -277,7 +277,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       false,
       topLevel,
       "baz",
-      []
+      [:]
       ), { AggregateMetric value ->
         value.getHitCount() == 1 && value.getTopLevelCount() == topLevelCount && value.getDuration() == 100
       })
@@ -333,7 +333,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       false,
       false,
       "baz",
-      []
+      [:]
       ), { AggregateMetric value ->
         value.getHitCount() == count && value.getDuration() == count * duration
       })
@@ -346,7 +346,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       false,
       false,
       "baz",
-      []
+      [:]
       ), { AggregateMetric value ->
         value.getHitCount() == count && value.getDuration() == count * duration * 2
       })
@@ -396,7 +396,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
         false,
         true,
         "baz",
-        []
+        [:]
         ), _) >> { MetricKey key, AggregateMetric value ->
           value.getHitCount() == 1 && value.getDuration() == duration
         }
@@ -410,7 +410,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       false,
       true,
       "baz",
-      []
+      [:]
       ), _)
     1 * writer.finishBucket() >> { latch.countDown() }
 
@@ -456,7 +456,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
         false,
         true,
         "baz",
-        []
+        [:]
         ), { AggregateMetric value ->
           value.getHitCount() == 1 && value.getDuration() == duration
         })
@@ -487,7 +487,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
         false,
         true,
         "baz",
-        []
+        [:]
         ),{ AggregateMetric value ->
           value.getHitCount() == 1 && value.getDuration() == duration
         })
@@ -501,7 +501,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
       false,
       true,
       "baz",
-      []
+      [:]
       ), _)
     1 * writer.finishBucket() >> { latch.countDown() }
 
@@ -547,7 +547,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
         false,
         true,
         "quux",
-        []
+        [:]
         ), { AggregateMetric value ->
           value.getHitCount() == 1 && value.getDuration() == duration
         })
@@ -603,7 +603,7 @@ class ConflatingMetricAggregatorTest extends DDSpecification {
         false,
         true,
         "garply",
-        []
+        [:]
         ), { AggregateMetric value ->
           value.getHitCount() == 1 && value.getDuration() == duration
         })

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

@@ -53,7 +53,7 @@ class SerializingMetricWriterTest extends DDSpecification {
         false,
         false,
         "client",
-        ["country:canada", "georegion:amer", "peer.service:remote-service"]
+        ["country":"canada", "georegion":"amer", "peer.service":"remote-service"]
         ),
         new AggregateMetric().recordDurations(10, new AtomicLongArray(1L))
         ),
@@ -67,7 +67,7 @@ class SerializingMetricWriterTest extends DDSpecification {
         true,
         false,
         "producer",
-        ["country:canada", "georegion:amer", "peer.service:remote-service"]
+        ["country":"canada", "georegion":"amer", "peer.service":"remote-service"]
         ),
         new AggregateMetric().recordDurations(9, new AtomicLongArray(1L))
         )
@@ -83,7 +83,7 @@ class SerializingMetricWriterTest extends DDSpecification {
           false,
           false,
           "producer",
-          ["messaging.destination" + i]
+          ["messaging.destination" : "dest" + i]
           ),
           new AggregateMetric().recordDurations(10, new AtomicLongArray(1L))
           )
@@ -175,9 +175,12 @@ class SerializingMetricWriterTest extends DDSpecification {
         ++elementCount
         assert unpacker.unpackString() == "PeerTags"
         int peerTagsLength = unpacker.unpackArrayHeader()
-        assert peerTagsLength == key.getPeerTags().length
+        assert peerTagsLength == key.getPeerTags().size()
         for (int i = 0; i < peerTagsLength; i++) {
-          assert unpacker.unpackString() == key.getPeerTags()[i] as String
+          def string = unpacker.unpackString()
+          def separatorPos = string.indexOf(':')
+          def tagVal = key.getPeerTags()[string.substring(0, separatorPos)]
+          assert tagVal == string.substring(separatorPos + 1)
         }
         ++elementCount
         assert unpacker.unpackString() == "Hits"

dd-trace-core/src/traceAgentTest/groovy/MetricsIntegrationTest.groovy

@@ -34,11 +34,11 @@ class MetricsIntegrationTest extends AbstractTraceAgentTest {
       )
     writer.startBucket(2, System.nanoTime(), SECONDS.toNanos(10))
     writer.add(
-      new MetricKey("resource1", "service1", "operation1", "sql", 0, false, true, "xyzzy", ["grault"]),
+      new MetricKey("resource1", "service1", "operation1", "sql", 0, false, true, "xyzzy", ["grault":"quux"]),
       new AggregateMetric().recordDurations(5, new AtomicLongArray(2, 1, 2, 250, 4, 5))
       )
     writer.add(
-      new MetricKey("resource2", "service2", "operation2", "web", 200, false, true, "xyzzy", ["grault"]),
+      new MetricKey("resource2", "service2", "operation2", "web", 200, false, true, "xyzzy", ["grault":"quux"]),
       new AggregateMetric().recordDurations(10, new AtomicLongArray(1, 1, 200, 2, 3, 4, 5, 6, 7, 8, 9))
       )
     writer.finishBucket()