Provide optimized writers for OpenTelemetry's "trace.proto" wire protocol

dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java

@@ -47,7 +47,6 @@ public final class ConfigDefaults {
   public static final boolean DEFAULT_STARTUP_LOGS_ENABLED = true;
 
   static final boolean DEFAULT_INJECT_DATADOG_ATTRIBUTE = true;
-  static final boolean DEFAULT_WRITER_BAGGAGE_INJECT = true;
   static final String DEFAULT_SITE = "datadoghq.com";
 
   static final boolean DEFAULT_CODE_ORIGIN_FOR_SPANS_INTERFACE_SUPPORT = false;

dd-trace-api/src/main/java/datadog/trace/api/config/TracerConfig.java

@@ -17,6 +17,7 @@ public final class TracerConfig {
   public static final String ID_GENERATION_STRATEGY = "id.generation.strategy";
   public static final String WRITER_TYPE = "writer.type";
   public static final String WRITER_BAGGAGE_INJECT = "writer.baggage.inject";
+  public static final String WRITER_LINKS_INJECT = "writer.links.inject";
 
   public static final String PRIORITIZATION_TYPE = "prioritization.type";
   public static final String TRACE_AGENT_URL = "trace.agent.url";

dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java

@@ -209,6 +209,7 @@ public static final CoreTracerBuilder builder() {
   private final TimeSource timeSource;
   private final ProfilingContextIntegration profilingContextIntegration;
   private final boolean injectBaggageAsTags;
+  private final boolean injectLinksAsTags;
   private final boolean flushOnClose;
   private final Collection<Runnable> shutdownListeners = new CopyOnWriteArrayList<>();
 
@@ -326,6 +327,7 @@ public static class CoreTracerBuilder {
     private boolean reportInTracerFlare;
     private boolean pollForTracingConfiguration;
     private boolean injectBaggageAsTags;
+    private boolean injectLinksAsTags;
     private boolean flushOnClose;
 
     public CoreTracerBuilder serviceName(String serviceName) {
@@ -471,6 +473,11 @@ public CoreTracerBuilder injectBaggageAsTags(boolean injectBaggageAsTags) {
       return this;
     }
 
+    public CoreTracerBuilder injectLinksAsTags(boolean injectLinksAsTags) {
+      this.injectLinksAsTags = injectLinksAsTags;
+      return this;
+    }
+
     public CoreTracerBuilder flushOnClose(boolean flushOnClose) {
       this.flushOnClose = flushOnClose;
       return this;
@@ -506,6 +513,7 @@ public CoreTracerBuilder config(final Config config) {
       partialFlushMinSpans(config.getPartialFlushMinSpans());
       strictTraceWrites(config.isTraceStrictWritesEnabled());
       injectBaggageAsTags(config.isInjectBaggageAsTagsEnabled());
+      injectLinksAsTags(config.isInjectLinksAsTagsEnabled());
       flushOnClose(config.isCiVisibilityEnabled());
       return this;
     }
@@ -538,6 +546,7 @@ public CoreTracer build() {
           reportInTracerFlare,
           pollForTracingConfiguration,
           injectBaggageAsTags,
+          injectLinksAsTags,
           flushOnClose);
     }
   }
@@ -569,6 +578,7 @@ private CoreTracer(
       final boolean reportInTracerFlare,
       final boolean pollForTracingConfiguration,
       final boolean injectBaggageAsTags,
+      final boolean injectLinksAsTags,
       final boolean flushOnClose) {
     this(
         config,
@@ -597,6 +607,7 @@ private CoreTracer(
         reportInTracerFlare,
         pollForTracingConfiguration,
         injectBaggageAsTags,
+        injectLinksAsTags,
         flushOnClose);
   }
 
@@ -628,6 +639,7 @@ private CoreTracer(
       final boolean reportInTracerFlare,
       final boolean pollForTracingConfiguration,
       final boolean injectBaggageAsTags,
+      final boolean injectLinksAsTags,
       final boolean flushOnClose) {
 
     assert localRootSpanTags != null;
@@ -864,6 +876,7 @@ private CoreTracer(
     propagationTagsFactory = PropagationTags.factory(config);
     this.profilingContextIntegration = profilingContextIntegration;
     this.injectBaggageAsTags = injectBaggageAsTags;
+    this.injectLinksAsTags = injectLinksAsTags;
     this.flushOnClose = flushOnClose;
     this.allowInferredServices = SpanNaming.instance().namingSchema().allowInferredServices();
     if (profilingContextIntegration != ProfilingContextIntegration.NoOp.INSTANCE) {
@@ -2119,7 +2132,8 @@ protected static final DDSpanContext buildSpanContext(
               tracer.disableSamplingMechanismValidation,
               propagationTags,
               tracer.profilingContextIntegration,
-              tracer.injectBaggageAsTags);
+              tracer.injectBaggageAsTags,
+              tracer.injectLinksAsTags);
 
       // By setting the tags on the context we apply decorators to any tags that have been set via
       // the builder. This is the order that the tags were added previously, but maybe the `tags`

dd-trace-core/src/main/java/datadog/trace/core/DDSpan.java

@@ -883,7 +883,7 @@ public TraceConfig traceConfig() {
     return context.getTraceCollector().getTraceConfig();
   }
 
-  List<? extends AgentSpanLink> getLinks() {
+  public List<? extends AgentSpanLink> getLinks() {
     return this.links;
   }
 

dd-trace-core/src/main/java/datadog/trace/core/DDSpanContext.java

@@ -184,6 +184,7 @@ public class DDSpanContext
 
   private final ProfilingContextIntegration profilingContextIntegration;
   private final boolean injectBaggageAsTags;
+  private final boolean injectLinksAsTags;
   private volatile int encodedOperationName;
   private volatile int encodedResourceName;
 
@@ -238,6 +239,7 @@ public DDSpanContext(
         disableSamplingMechanismValidation,
         propagationTags,
         ProfilingContextIntegration.NoOp.INSTANCE,
+        true,
         true);
   }
 
@@ -261,7 +263,8 @@ public DDSpanContext(
       final PathwayContext pathwayContext,
       final boolean disableSamplingMechanismValidation,
       final PropagationTags propagationTags,
-      final boolean injectBaggageAsTags) {
+      final boolean injectBaggageAsTags,
+      final boolean injectLinksAsTags) {
     this(
         traceId,
         spanId,
@@ -286,7 +289,8 @@ public DDSpanContext(
         disableSamplingMechanismValidation,
         propagationTags,
         ProfilingContextIntegration.NoOp.INSTANCE,
-        injectBaggageAsTags);
+        injectBaggageAsTags,
+        injectLinksAsTags);
   }
 
   public DDSpanContext(
@@ -313,7 +317,8 @@ public DDSpanContext(
       final boolean disableSamplingMechanismValidation,
       final PropagationTags propagationTags,
       final ProfilingContextIntegration profilingContextIntegration,
-      final boolean injectBaggageAsTags) {
+      final boolean injectBaggageAsTags,
+      final boolean injectLinksAsTags) {
 
     assert traceCollector != null;
     this.traceCollector = traceCollector;
@@ -370,6 +375,7 @@ public DDSpanContext(
             : traceCollector.getTracer().getPropagationTagsFactory().empty();
     this.propagationTags.updateTraceIdHighOrderBits(this.traceId.toHighOrderLong());
     this.injectBaggageAsTags = injectBaggageAsTags;
+    this.injectLinksAsTags = injectLinksAsTags;
     if (origin != null) {
       setOrigin(origin);
     }
@@ -1177,10 +1183,14 @@ void processTagsAndBaggage(
       // Tags
       TagsPostProcessorFactory.lazyProcessor().processTags(unsafeTags, this, restrictedSpan);
 
-      String linksTag = DDSpanLink.toTag(restrictedSpan.getLinks());
-      if (linksTag != null) {
-        unsafeTags.put(SPAN_LINKS, linksTag);
+      // Links
+      if (injectLinksAsTags) {
+        String linksTag = DDSpanLink.toTag(restrictedSpan.getLinks());
+        if (linksTag != null) {
+          unsafeTags.set(SPAN_LINKS, linksTag);
+        }
       }
+
       // Baggage
       Map<String, String> baggageItemsWithPropagationTags;
       if (injectBaggageAsTags) {

dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpCommonProto.java

@@ -15,6 +15,7 @@
 import datadog.communication.serialization.SimpleUtf8Cache;
 import datadog.communication.serialization.StreamingBuffer;
 import datadog.trace.api.Config;
+import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
 import datadog.trace.bootstrap.otel.common.OtelInstrumentationScope;
 import java.nio.ByteBuffer;
 import java.util.List;
@@ -45,6 +46,12 @@ private OtlpCommonProto() {}
           ? new GenerationalUtf8Cache(Config.get().getTagValueUtf8CacheSize())
           : null;
 
+  public static void recalibrateCaches() {
+    if (VALUE_CACHE != null) {
+      VALUE_CACHE.recalibrate();
+    }
+  }
+
   public static int sizeVarInt(int value) {
     return 1 + (31 - Integer.numberOfLeadingZeros(value)) / 7;
   }
@@ -153,8 +160,13 @@ public static void writeInstrumentationScope(
   }
 
   @SuppressWarnings("unchecked")
-  public static void writeAttribute(StreamingBuffer buf, int type, String key, Object value) {
-    byte[] keyUtf8 = keyUtf8(key);
+  public static void writeAttribute(StreamingBuffer buf, int type, CharSequence key, Object value) {
+    byte[] keyUtf8;
+    if (key instanceof UTF8BytesString) {
+      keyUtf8 = ((UTF8BytesString) key).getUtf8Bytes();
+    } else {
+      keyUtf8 = keyUtf8(key.toString());
+    }
     switch (type) {
       case STRING:
         writeStringAttribute(buf, keyUtf8, valueUtf8((String) value));
@@ -163,10 +175,10 @@ public static void writeAttribute(StreamingBuffer buf, int type, String key, Obj
         writeBooleanAttribute(buf, keyUtf8, (boolean) value);
         break;
       case LONG:
-        writeLongAttribute(buf, keyUtf8, (long) value);
+        writeLongAttribute(buf, keyUtf8, ((Number) value).longValue());
         break;
       case DOUBLE:
-        writeDoubleAttribute(buf, keyUtf8, (double) value);
+        writeDoubleAttribute(buf, keyUtf8, ((Number) value).doubleValue());
         break;
       case STRING_ARRAY:
         writeStringArrayAttribute(buf, keyUtf8, (List<String>) value);
@@ -175,16 +187,29 @@ public static void writeAttribute(StreamingBuffer buf, int type, String key, Obj
         writeBooleanArrayAttribute(buf, keyUtf8, (List<Boolean>) value);
         break;
       case LONG_ARRAY:
-        writeLongArrayAttribute(buf, keyUtf8, (List<Long>) value);
+        writeLongArrayAttribute(buf, keyUtf8, (List<? extends Number>) value);
         break;
       case DOUBLE_ARRAY:
-        writeDoubleArrayAttribute(buf, keyUtf8, (List<Double>) value);
+        writeDoubleArrayAttribute(buf, keyUtf8, (List<? extends Number>) value);
         break;
       default:
         throw new IllegalArgumentException("Unknown attribute type: " + type);
     }
   }
 
+  public static void writeAttribute(
+      StreamingBuffer buf, UTF8BytesString key, UTF8BytesString value) {
+    writeStringAttribute(buf, key.getUtf8Bytes(), value.getUtf8Bytes());
+  }
+
+  public static void writeAttribute(StreamingBuffer buf, UTF8BytesString key, String value) {
+    writeStringAttribute(buf, key.getUtf8Bytes(), valueUtf8(value));
+  }
+
+  public static void writeAttribute(StreamingBuffer buf, UTF8BytesString key, long value) {
+    writeLongAttribute(buf, key.getUtf8Bytes(), value);
+  }
+
   private static byte[] keyUtf8(String key) {
     return KEY_CACHE != null ? KEY_CACHE.getUtf8(key) : key.getBytes(UTF_8);
   }
@@ -301,10 +326,10 @@ private static void writeBooleanArrayAttribute(
   }
 
   private static void writeLongArrayAttribute(
-      StreamingBuffer buf, byte[] keyUtf8, List<Long> values) {
+      StreamingBuffer buf, byte[] keyUtf8, List<? extends Number> values) {
     long[] longValues = new long[values.size()];
     for (int i = 0; i < longValues.length; i++) {
-      longValues[i] = values.get(i); // avoid repeated unboxing later
+      longValues[i] = values.get(i).longValue(); // avoid repeated unboxing later
     }
     int arraySize = 0;
     for (long longValue : longValues) {
@@ -332,7 +357,7 @@ private static void writeLongArrayAttribute(
   }
 
   private static void writeDoubleArrayAttribute(
-      StreamingBuffer buf, byte[] keyUtf8, List<Double> values) {
+      StreamingBuffer buf, byte[] keyUtf8, List<? extends Number> values) {
     int arraySize = 11 * values.size();
     int valueSize = 1 + sizeVarInt(arraySize) + arraySize;
     int keyValueSize =
@@ -345,11 +370,11 @@ private static void writeDoubleArrayAttribute(
     writeVarInt(buf, valueSize);
     writeTag(buf, 5, LEN_WIRE_TYPE);
     writeVarInt(buf, arraySize);
-    for (double value : values) {
+    for (Number value : values) {
       writeTag(buf, 1, LEN_WIRE_TYPE);
       buf.put((byte) 9);
       writeTag(buf, 4, I64_WIRE_TYPE);
-      writeI64(buf, value);
+      writeI64(buf, value.doubleValue());
     }
   }
 }