Telemetry for OTLP traces/metrics/logs (#12057)

Telemetry for OTLP traces/metrics/logs:

* Datadog tracer health metrics
* otel.metrics_export_{attempts,successes,failures}
* otel.log_records

OTLP telemetry metrics are tagged by protocol and encoding (per signal)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Remove HealthMetrics for OTLP traces (since HealthMetrics relies on StatsD client) and use telemetry instead

Simplify OTLP telemetry

Stage telemetry metrics separate to collection

Co-authored-by: stuart.mcculloch <stuart.mcculloch@datadoghq.com>

Author: mcculls

dd-trace-core/src/main/java/datadog/trace/common/writer/OtlpPayloadDispatcher.java

@@ -1,5 +1,6 @@
 package datadog.trace.common.writer;
 
+import datadog.trace.api.telemetry.OtlpTelemetry;
 import datadog.trace.core.CoreSpan;
 import datadog.trace.core.otlp.common.OtlpPayload;
 import datadog.trace.core.otlp.common.OtlpSender;
@@ -37,7 +38,9 @@ public void flush() {
     try {
       OtlpPayload payload = collector.collectTraces();
       if (payload != OtlpPayload.EMPTY) {
-        sender.send(payload);
+        OtlpTelemetry.getInstance().onTracesExportAttempt();
+        RemoteApi.Response response = sender.send(payload);
+        OtlpTelemetry.getInstance().onTracesExportComplete(response.success());
       }
     } catch (RuntimeException e) { // don't catch severe Errors
       log.debug("Failed to send OTLP payload", e);
@@ -46,7 +49,7 @@ public void flush() {
 
   @Override
   public void onDroppedTrace(int spanCount) {
-    // TODO: surface drop counts via HealthMetrics
+    // no telemetry currently tracked for dropped traces
   }
 
   @Override

dd-trace-core/src/main/java/datadog/trace/common/writer/OtlpWriter.java

@@ -38,11 +38,10 @@ public static OtlpWriterBuilder builder() {
       TraceProcessingWorker worker,
       PayloadDispatcher dispatcher,
       OtlpSender sender,
-      HealthMetrics healthMetrics,
       int flushTimeout,
       TimeUnit flushTimeoutUnit,
       boolean alwaysFlush) {
-    super(worker, dispatcher, healthMetrics, flushTimeout, flushTimeoutUnit, alwaysFlush);
+    super(worker, dispatcher, HealthMetrics.NO_OP, flushTimeout, flushTimeoutUnit, alwaysFlush);
     this.sender = sender;
   }
 
@@ -64,7 +63,6 @@ public static class OtlpWriterBuilder {
     private OtlpConfig.Protocol protocol = OtlpConfig.Protocol.HTTP_PROTOBUF;
     private OtlpConfig.Compression compression = OtlpConfig.Compression.NONE;
     private int traceBufferSize = BUFFER_SIZE;
-    private HealthMetrics healthMetrics = HealthMetrics.NO_OP;
     private int flushIntervalMilliseconds = 1000;
     private int flushTimeout = 1;
     private TimeUnit flushTimeoutUnit = TimeUnit.SECONDS;
@@ -102,11 +100,6 @@ public OtlpWriterBuilder traceBufferSize(int traceBufferSize) {
       return this;
     }
 
-    public OtlpWriterBuilder healthMetrics(HealthMetrics healthMetrics) {
-      this.healthMetrics = healthMetrics;
-      return this;
-    }
-
     public OtlpWriterBuilder flushIntervalMilliseconds(int flushIntervalMilliseconds) {
       this.flushIntervalMilliseconds = flushIntervalMilliseconds;
       return this;
@@ -151,7 +144,7 @@ public OtlpWriter build() {
       final TraceProcessingWorker worker =
           new TraceProcessingWorker(
               traceBufferSize,
-              healthMetrics,
+              HealthMetrics.NO_OP,
               dispatcher,
               DroppingPolicy.DISABLED,
               Prioritization.FAST_LANE,
@@ -160,7 +153,7 @@ public OtlpWriter build() {
               singleSpanSampler);
 
       return new OtlpWriter(
-          worker, dispatcher, sender, healthMetrics, flushTimeout, flushTimeoutUnit, alwaysFlush);
+          worker, dispatcher, sender, flushTimeout, flushTimeoutUnit, alwaysFlush);
     }
   }
 }

dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java

@@ -76,7 +76,6 @@ public static Writer createWriter(
           .protocol(config.getOtlpTracesProtocol())
           .compression(config.getOtlpTracesCompression())
           .timeoutMillis(config.getOtlpTracesTimeout())
-          .healthMetrics(healthMetrics)
           .spanSamplingRules(singleSpanSampler)
           .flushIntervalMilliseconds(flushIntervalMilliseconds)
           .build();

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

@@ -2,18 +2,16 @@
 
 import static datadog.communication.http.OkHttpUtils.buildHttp2Client;
 import static datadog.communication.http.OkHttpUtils.isPlainHttp;
-import static datadog.communication.http.OkHttpUtils.sendWithRetries;
 
 import datadog.communication.http.HttpRetryPolicy;
 import datadog.logging.RatelimitedLogger;
 import datadog.trace.api.config.OtlpConfig.Compression;
-import java.io.IOException;
+import datadog.trace.common.writer.RemoteApi;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import okhttp3.HttpUrl;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
-import okhttp3.Response;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,20 +53,8 @@ public OtlpGrpcSender(
   }
 
   @Override
-  public void send(OtlpPayload payload) {
-    Request request = makeRequest(payload);
-    try (Response response = sendWithRetries(client, retryPolicy, request)) {
-      if (!response.isSuccessful()) {
-        RATELIMITED_LOGGER.warn(
-            "OTLP export to {} failed with status {}: {}",
-            request.url(),
-            response.code(),
-            response.message());
-      }
-    } catch (IOException e) {
-      RATELIMITED_LOGGER.warn(
-          "OTLP export to {} failed with exception: {}", request.url(), e.toString());
-    }
+  public RemoteApi.Response send(OtlpPayload payload) {
+    return OtlpSenderSupport.send(client, retryPolicy, makeRequest(payload), RATELIMITED_LOGGER);
   }
 
   @Override

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

@@ -2,18 +2,16 @@
 
 import static datadog.communication.http.OkHttpUtils.buildHttpClient;
 import static datadog.communication.http.OkHttpUtils.isPlainHttp;
-import static datadog.communication.http.OkHttpUtils.sendWithRetries;
 
 import datadog.communication.http.HttpRetryPolicy;
 import datadog.logging.RatelimitedLogger;
 import datadog.trace.api.config.OtlpConfig.Compression;
-import java.io.IOException;
+import datadog.trace.common.writer.RemoteApi;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import okhttp3.HttpUrl;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
-import okhttp3.Response;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -59,20 +57,8 @@ public HttpUrl url() {
   }
 
   @Override
-  public void send(OtlpPayload payload) {
-    Request request = makeRequest(payload);
-    try (Response response = sendWithRetries(client, retryPolicy, request)) {
-      if (!response.isSuccessful()) {
-        RATELIMITED_LOGGER.warn(
-            "OTLP export to {} failed with status {}: {}",
-            request.url(),
-            response.code(),
-            response.message());
-      }
-    } catch (IOException e) {
-      RATELIMITED_LOGGER.warn(
-          "OTLP export to {} failed with exception: {}", request.url(), e.toString());
-    }
+  public RemoteApi.Response send(OtlpPayload payload) {
+    return OtlpSenderSupport.send(client, retryPolicy, makeRequest(payload), RATELIMITED_LOGGER);
   }
 
   @Override

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

@@ -1,8 +1,10 @@
 package datadog.trace.core.otlp.common;
 
+import datadog.trace.common.writer.RemoteApi;
+
 /** Sends chunks of OTLP data. */
 public interface OtlpSender {
-  void send(OtlpPayload payload);
+  RemoteApi.Response send(OtlpPayload payload);
 
   void shutdown();
 }

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

@@ -0,0 +1,39 @@
+package datadog.trace.core.otlp.common;
+
+import static datadog.communication.http.OkHttpUtils.sendWithRetries;
+import static datadog.trace.common.writer.RemoteApi.Response.failed;
+import static datadog.trace.common.writer.RemoteApi.Response.success;
+
+import datadog.communication.http.HttpRetryPolicy;
+import datadog.logging.RatelimitedLogger;
+import datadog.trace.common.writer.RemoteApi;
+import java.io.IOException;
+import okhttp3.OkHttpClient;
+
+/** Shared request execution and response handling for {@link OtlpSender} implementations. */
+final class OtlpSenderSupport {
+  private OtlpSenderSupport() {}
+
+  /** Executes the given request with retries, logging failures via the rate-limited logger. */
+  static RemoteApi.Response send(
+      OkHttpClient client,
+      HttpRetryPolicy.Factory retryPolicy,
+      okhttp3.Request request,
+      RatelimitedLogger ratelimitedLogger) {
+    try (okhttp3.Response response = sendWithRetries(client, retryPolicy, request)) {
+      if (response.isSuccessful()) {
+        return success(response.code());
+      }
+      ratelimitedLogger.warn(
+          "OTLP export to {} failed with status {}: {}",
+          request.url(),
+          response.code(),
+          response.message());
+      return failed(response.code());
+    } catch (IOException e) {
+      ratelimitedLogger.warn(
+          "OTLP export to {} failed with exception: {}", request.url(), e.toString());
+      return failed(e);
+    }
+  }
+}

dd-trace-core/src/main/java/datadog/trace/core/otlp/logs/OtlpLogsCollector.java

@@ -7,4 +7,7 @@ public abstract class OtlpLogsCollector {
 
   /** Waits for logs to be batched within the given interval. */
   public abstract OtlpPayload waitForLogs(int intervalMillis);
+
+  /** Number of log records collected. */
+  public abstract int getLogRecordCount();
 }

dd-trace-core/src/main/java/datadog/trace/core/otlp/logs/OtlpLogsJsonCollector.java

@@ -36,6 +36,7 @@ public final class OtlpLogsJsonCollector extends OtlpLogsCollector
   private JsonWriter writer;
   private boolean anyLogRecordWritten;
   private boolean logRecordStarted;
+  private int logRecordCount;
 
   private final LazyJsonArray attributesArray = new LazyJsonArray();
 
@@ -65,6 +66,8 @@ OtlpPayload collectLogs(ObjIntConsumer<OtlpLogsVisitor> processor, int intervalM
 
   /** Prepare temporary elements to collect logs data. */
   private void start() {
+    logRecordCount = 0;
+
     writer = new JsonWriter();
     writer.beginObject();
     writer.name("resourceLogs").beginArray();
@@ -86,6 +89,11 @@ private void stop() {
     currentScope = null;
   }
 
+  @Override
+  public int getLogRecordCount() {
+    return logRecordCount;
+  }
+
   @Override
   public OtlpScopedLogsVisitor visitScopedLogs(OtelInstrumentationScope scope) {
     if (currentScope != null) {
@@ -119,6 +127,7 @@ public void visitLogRecord(OtlpLogRecord logRecord) {
 
     logRecordStarted = false;
     anyLogRecordWritten = true;
+    logRecordCount++;
   }
 
   // opens the log record object on first attribute or value written for it

dd-trace-core/src/main/java/datadog/trace/core/otlp/logs/OtlpLogsProtoCollector.java

@@ -41,6 +41,7 @@ public final class OtlpLogsProtoCollector extends OtlpLogsCollector
   // total number of chunked bytes at different nesting levels
   private int payloadBytes;
   private int scopedBytes;
+  private int logRecordCount;
 
   private OtelInstrumentationScope currentScope;
 
@@ -68,6 +69,7 @@ OtlpPayload collectLogs(ObjIntConsumer<OtlpLogsVisitor> processor, int intervalM
 
   /** Prepare temporary elements to collect logs data. */
   private void start() {
+    logRecordCount = 0;
 
     // remove stale entries from caches
     OtlpCommonProto.recalibrateCaches();
@@ -84,6 +86,11 @@ private void stop() {
     currentScope = null;
   }
 
+  @Override
+  public int getLogRecordCount() {
+    return logRecordCount;
+  }
+
   @Override
   public OtlpScopedLogsVisitor visitScopedLogs(OtelInstrumentationScope scope) {
     if (currentScope != null) {
@@ -96,6 +103,7 @@ public OtlpScopedLogsVisitor visitScopedLogs(OtelInstrumentationScope scope) {
   @Override
   public void visitLogRecord(OtlpLogRecord logRecord) {
     scopedBytes += recordLogRecordMessage(buf, logRecord, protobuf);
+    logRecordCount++;
   }
 
   @Override