Cleanup OTLP code

mcculls · mcculls · commit 7d10c73bdc88 · 2026-04-07T02:53:21.000+01:00
diff --git a/dd-java-agent/agent-otel/otel-bootstrap/src/main/java/datadog/trace/bootstrap/otel/metrics/OtelInstrumentBuilder.java b/dd-java-agent/agent-otel/otel-bootstrap/src/main/java/datadog/trace/bootstrap/otel/metrics/OtelInstrumentBuilder.java
@@ -1,5 +1,9 @@
 package datadog.trace.bootstrap.otel.metrics;
 
+import static datadog.trace.bootstrap.otel.metrics.OtelInstrumentType.OBSERVABLE_COUNTER;
+import static datadog.trace.bootstrap.otel.metrics.OtelInstrumentType.OBSERVABLE_GAUGE;
+import static datadog.trace.bootstrap.otel.metrics.OtelInstrumentType.OBSERVABLE_UP_DOWN_COUNTER;
+
 import javax.annotation.Nullable;
 
 public final class OtelInstrumentBuilder {
@@ -91,11 +95,11 @@ public OtelInstrumentDescriptor observableDescriptor() {
   private OtelInstrumentType observableType(OtelInstrumentType instrumentType) {
     switch (instrumentType) {
       case COUNTER:
-        return OtelInstrumentType.OBSERVABLE_COUNTER;
+        return OBSERVABLE_COUNTER;
       case UP_DOWN_COUNTER:
-        return OtelInstrumentType.OBSERVABLE_UP_DOWN_COUNTER;
+        return OBSERVABLE_UP_DOWN_COUNTER;
       case GAUGE:
-        return OtelInstrumentType.OBSERVABLE_GAUGE;
+        return OBSERVABLE_GAUGE;
       default:
         throw new IllegalArgumentException(instrumentType + " has no observable equivalent");
     }
diff --git a/dd-java-agent/agent-otel/otel-bootstrap/src/main/java/datadog/trace/bootstrap/otel/metrics/data/OtelMetricStorage.java b/dd-java-agent/agent-otel/otel-bootstrap/src/main/java/datadog/trace/bootstrap/otel/metrics/data/OtelMetricStorage.java
@@ -1,5 +1,9 @@
 package datadog.trace.bootstrap.otel.metrics.data;
 
+import static datadog.trace.bootstrap.otel.metrics.OtelInstrumentType.COUNTER;
+import static datadog.trace.bootstrap.otel.metrics.OtelInstrumentType.HISTOGRAM;
+import static datadog.trace.bootstrap.otel.metrics.OtelInstrumentType.OBSERVABLE_COUNTER;
+
 import datadog.logging.RatelimitedLogger;
 import datadog.trace.api.Config;
 import datadog.trace.api.config.OtlpConfig;
@@ -63,12 +67,10 @@ private static boolean shouldResetOnCollect(OtelInstrumentType type) {
     switch (TEMPORALITY_PREFERENCE) {
       case DELTA:
         // gauges and up/down counters stay as cumulative
-        return type == OtelInstrumentType.HISTOGRAM
-            || type == OtelInstrumentType.COUNTER
-            || type == OtelInstrumentType.OBSERVABLE_COUNTER;
+        return type == HISTOGRAM || type == COUNTER || type == OBSERVABLE_COUNTER;
       case LOWMEMORY:
         // observable counters, gauges, and up/down counters stay as cumulative
-        return type == OtelInstrumentType.HISTOGRAM || type == OtelInstrumentType.COUNTER;
+        return type == HISTOGRAM || type == COUNTER;
       case CUMULATIVE:
       default:
         return false;
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java b/dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java
@@ -1418,6 +1418,9 @@ public void close() {
     RumInjector.shutdownTelemetry();
     AgentMeter.statsDClient().close();
     metricsAggregator.close();
+    if (initialConfig.isMetricsOtlpExporterEnabled()) {
+      OtlpMetricsService.INSTANCE.shutdown();
+    }
     dataStreamsMonitoring.close();
     externalAgentLauncher.close();
     healthMetrics.close();
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpCommonProto.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpCommonProto.java
@@ -1,5 +1,13 @@
 package datadog.trace.core.otlp.common;
 
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.BOOLEAN;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.BOOLEAN_ARRAY;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.DOUBLE;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.DOUBLE_ARRAY;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.LONG;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.LONG_ARRAY;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ARRAY;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
 import datadog.communication.serialization.GenerationalUtf8Cache;
@@ -8,7 +16,6 @@
 import datadog.communication.serialization.StreamingBuffer;
 import datadog.trace.api.Config;
 import datadog.trace.bootstrap.otel.common.OtelInstrumentationScope;
-import datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor;
 import java.nio.ByteBuffer;
 import java.util.List;
 
@@ -149,28 +156,28 @@ public static void writeInstrumentationScope(
   public static void writeAttribute(StreamingBuffer buf, int type, String key, Object value) {
     byte[] keyUtf8 = keyUtf8(key);
     switch (type) {
-      case OtlpAttributeVisitor.STRING:
+      case STRING:
         writeStringAttribute(buf, keyUtf8, valueUtf8((String) value));
         break;
-      case OtlpAttributeVisitor.BOOLEAN:
+      case BOOLEAN:
         writeBooleanAttribute(buf, keyUtf8, (boolean) value);
         break;
-      case OtlpAttributeVisitor.LONG:
+      case LONG:
         writeLongAttribute(buf, keyUtf8, (long) value);
         break;
-      case OtlpAttributeVisitor.DOUBLE:
+      case DOUBLE:
         writeDoubleAttribute(buf, keyUtf8, (double) value);
         break;
-      case OtlpAttributeVisitor.STRING_ARRAY:
+      case STRING_ARRAY:
         writeStringArrayAttribute(buf, keyUtf8, (List<String>) value);
         break;
-      case OtlpAttributeVisitor.BOOLEAN_ARRAY:
+      case BOOLEAN_ARRAY:
         writeBooleanArrayAttribute(buf, keyUtf8, (List<Boolean>) value);
         break;
-      case OtlpAttributeVisitor.LONG_ARRAY:
+      case LONG_ARRAY:
         writeLongArrayAttribute(buf, keyUtf8, (List<Long>) value);
         break;
-      case OtlpAttributeVisitor.DOUBLE_ARRAY:
+      case DOUBLE_ARRAY:
         writeDoubleArrayAttribute(buf, keyUtf8, (List<Double>) value);
         break;
       default:
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpHttpRequestBody.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpHttpRequestBody.java
@@ -5,21 +5,25 @@
 import okhttp3.MediaType;
 import okhttp3.RequestBody;
 import okio.BufferedSink;
+import okio.GzipSink;
+import okio.Okio;
 
 /** Wraps an {@link OtlpPayload} as an OkHttp {@link RequestBody}. */
 public final class OtlpHttpRequestBody extends RequestBody {
 
   private final OtlpPayload payload;
   private final MediaType mediaType;
+  private final boolean gzip;
 
-  public OtlpHttpRequestBody(OtlpPayload payload) {
+  public OtlpHttpRequestBody(OtlpPayload payload, boolean gzip) {
     this.payload = payload;
     this.mediaType = MediaType.get(payload.getContentType());
+    this.gzip = gzip;
   }
 
   @Override
   public long contentLength() {
-    return payload.getContentLength();
+    return gzip ? -1 : payload.getContentLength();
   }
 
   @Override
@@ -29,6 +33,12 @@ public MediaType contentType() {
 
   @Override
   public void writeTo(@Nonnull BufferedSink sink) throws IOException {
-    payload.drain(sink::write);
+    if (gzip) {
+      try (BufferedSink gzipSink = Okio.buffer(new GzipSink(sink))) {
+        payload.drain(gzipSink::write);
+      }
+    } else {
+      payload.drain(sink::write);
+    }
   }
 }
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpHttpSender.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpHttpSender.java
@@ -1,7 +1,6 @@
 package datadog.trace.core.otlp.common;
 
 import static datadog.communication.http.OkHttpUtils.buildHttpClient;
-import static datadog.communication.http.OkHttpUtils.gzippedRequestBodyOf;
 import static datadog.communication.http.OkHttpUtils.isPlainHttp;
 import static datadog.communication.http.OkHttpUtils.sendWithRetries;
 
@@ -14,7 +13,6 @@
 import okhttp3.HttpUrl;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
-import okhttp3.RequestBody;
 import okhttp3.Response;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -56,10 +54,8 @@ public OtlpHttpSender(
     this.client = buildHttpClient(isPlainHttp(url), unixDomainSocketPath, null, timeoutMillis);
   }
 
+  @Override
   public void send(OtlpPayload payload) {
-    if (payload == OtlpPayload.EMPTY) {
-      return; // nothing to send
-    }
     Request request = makeRequest(payload);
     try (Response response = sendWithRetries(client, retryPolicy, request)) {
       if (!response.isSuccessful()) {
@@ -75,30 +71,20 @@ public void send(OtlpPayload payload) {
     }
   }
 
+  @Override
   public void shutdown() {
     client.connectionPool().evictAll();
   }
 
   private Request makeRequest(OtlpPayload payload) {
-    Request.Builder requestBuilder =
-        new Request.Builder().url(url).header("Content-Type", payload.getContentType());
-
+    Request.Builder requestBuilder = new Request.Builder().url(url);
     if (gzip) {
-      requestBuilder
-          .header("Content-Length", "-1")
-          .header("Content-Encoding", "gzip")
-          .header("Transfer-Encoding", "chunked");
-    } else {
-      requestBuilder.header("Content-Length", String.valueOf(payload.getContentLength()));
+      requestBuilder.header("Content-Encoding", "gzip").header("Transfer-Encoding", "chunked");
     }
 
+    // add configured headers to the request
     headers.forEach(requestBuilder::addHeader);
 
-    RequestBody requestBody = new OtlpHttpRequestBody(payload);
-    if (gzip) {
-      requestBody = gzippedRequestBodyOf(requestBody);
-    }
-
-    return requestBuilder.post(requestBody).build();
+    return requestBuilder.post(new OtlpHttpRequestBody(payload, gzip)).build();
   }
 }
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java
@@ -18,8 +18,6 @@
 public final class OtlpResourceProto {
   private OtlpResourceProto() {}
 
-  public static final byte[] RESOURCE_MESSAGE = buildResourceMessage(Config.get());
-
   private static final Set<String> IGNORED_GLOBAL_TAGS =
       new HashSet<>(
           Arrays.asList(
@@ -30,6 +28,8 @@ private OtlpResourceProto() {}
               "deployment.environment.name",
               "service.version"));
 
+  public static final byte[] RESOURCE_MESSAGE = buildResourceMessage(Config.get());
+
   static byte[] buildResourceMessage(Config config) {
     GrowableBuffer buf = new GrowableBuffer(512);
 
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpSender.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpSender.java
@@ -3,4 +3,6 @@
 /** Sends chunks of OTLP data. */
 public interface OtlpSender {
   void send(OtlpPayload payload);
+
+  void shutdown();
 }
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsCollector.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsCollector.java
@@ -4,5 +4,7 @@
 
 /** Collects metrics ready for export. */
 public interface OtlpMetricsCollector {
+  OtlpMetricsCollector NOOP_COLLECTOR = () -> OtlpPayload.EMPTY;
+
   OtlpPayload collectMetrics();
 }
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsService.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsService.java
@@ -4,12 +4,17 @@
 
 import datadog.trace.api.Config;
 import datadog.trace.core.otlp.common.OtlpHttpSender;
+import datadog.trace.core.otlp.common.OtlpPayload;
 import datadog.trace.core.otlp.common.OtlpSender;
 import datadog.trace.util.AgentTaskScheduler;
 import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /** Periodic service to collect OpenTelemetry metrics and export them over OTLP. */
 public final class OtlpMetricsService {
+  private static final Logger LOGGER = LoggerFactory.getLogger(OtlpMetricsService.class);
+
   public static final OtlpMetricsService INSTANCE = new OtlpMetricsService(Config.get());
 
   private final AgentTaskScheduler scheduler;
@@ -20,14 +25,23 @@ public final class OtlpMetricsService {
 
   private OtlpMetricsService(Config config) {
     this.scheduler = new AgentTaskScheduler(OTLP_METRICS_EXPORTER);
-    this.collector = OtlpMetricsProtoCollector.INSTANCE;
-    this.sender =
-        new OtlpHttpSender(
-            config.getOtlpMetricsEndpoint(),
-            "/v1/metrics",
-            config.getOtlpMetricsHeaders(),
-            config.getOtlpMetricsTimeout(),
-            config.getOtlpMetricsCompression());
+
+    switch (config.getOtlpMetricsProtocol()) {
+      case HTTP_PROTOBUF:
+        this.collector = OtlpMetricsProtoCollector.INSTANCE;
+        this.sender =
+            new OtlpHttpSender(
+                config.getOtlpMetricsEndpoint(),
+                "/v1/metrics",
+                config.getOtlpMetricsHeaders(),
+                config.getOtlpMetricsTimeout(),
+                config.getOtlpMetricsCompression());
+        break;
+      default:
+        LOGGER.debug("Unsupported OTLP metrics protocol: {}", config.getOtlpMetricsProtocol());
+        this.collector = OtlpMetricsCollector.NOOP_COLLECTOR;
+        this.sender = null;
+    }
 
     this.intervalMillis = config.getMetricsOtelInterval();
   }
@@ -41,7 +55,16 @@ public void flush() {
     scheduler.execute(this::export);
   }
 
+  public void shutdown() {
+    if (sender != null) {
+      sender.shutdown();
+    }
+  }
+
   private void export() {
-    sender.send(collector.collectMetrics());
+    OtlpPayload payload = collector.collectMetrics();
+    if (payload != OtlpPayload.EMPTY) {
+      sender.send(payload);
+    }
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -3,4 +3,6 @@`
`3`	`3`	`/** Sends chunks of OTLP data. */`
`4`	`4`	`public interface OtlpSender {`
`5`	`5`	`void send(OtlpPayload payload);`
	`6`	`+`
	`7`	`+ void shutdown();`
`6`	`8`	`}`
Original file line number	Diff line number	Diff line change
`@@ -4,5 +4,7 @@`
`4`	`4`
`5`	`5`	`/** Collects metrics ready for export. */`
`6`	`6`	`public interface OtlpMetricsCollector {`
	`7`	`+ OtlpMetricsCollector NOOP_COLLECTOR = () -> OtlpPayload.EMPTY;`
	`8`	`+`
`7`	`9`	`OtlpPayload collectMetrics();`
`8`	`10`	`}`