refactor: Extract OTLP sender infrastructure to communication/otlp-exporter

.github/CODEOWNERS

@@ -18,7 +18,7 @@
 
 # Shared communication
 /communication/*.*                                                   @DataDog/apm-java
-/communication/otlp-exporter/                                        @DataDog/apm-sdk-capabilities-java
+/communication/otlp-exporter/                                       @DataDog/apm-sdk-capabilities-java
 /communication/src/*/*/datadog/communication/*.*                     @DataDog/apm-java
 /communication/src/*/*/datadog/communication/ddagent/                @DataDog/apm-java
 /communication/src/*/*/datadog/communication/http/                   @DataDog/apm-java

communication/otlp-exporter/src/main/java/datadog/communication/otlp/OtlpCanonicalResourceAttributes.java

@@ -0,0 +1,40 @@
+package datadog.communication.otlp;
+
+import static datadog.communication.ddagent.TracerVersion.TRACER_VERSION;
+
+import datadog.trace.api.Config;
+import java.util.function.BiConsumer;
+
+/**
+ * The resource attributes every OTLP export reports. Shared by the tracer and the profiler so both
+ * describe the same producer, and kept in one place so an attribute is added exactly once.
+ */
+public final class OtlpCanonicalResourceAttributes {
+  private OtlpCanonicalResourceAttributes() {}
+
+  /**
+   * Visits {@code service.name}, {@code deployment.environment.name} and {@code service.version}
+   * (when not empty), {@code host.name} (when reported and not empty) and the {@code
+   * telemetry.sdk.*} attributes, in that order.
+   */
+  public static void visit(Config config, BiConsumer<String, String> visitor) {
+    visitor.accept("service.name", config.getServiceName());
+    String env = config.getEnv();
+    if (!env.isEmpty()) {
+      visitor.accept("deployment.environment.name", env);
+    }
+    String version = config.getVersion();
+    if (!version.isEmpty()) {
+      visitor.accept("service.version", version);
+    }
+    if (config.isReportHostName()) {
+      String hostName = config.getHostName();
+      if (hostName != null && !hostName.isEmpty()) {
+        visitor.accept("host.name", hostName);
+      }
+    }
+    visitor.accept("telemetry.sdk.name", "datadog");
+    visitor.accept("telemetry.sdk.version", TRACER_VERSION);
+    visitor.accept("telemetry.sdk.language", "java");
+  }
+}

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

@@ -1,14 +1,11 @@
-package datadog.trace.core.otlp.common;
+package datadog.communication.otlp;
 
 import static datadog.communication.http.OkHttpUtils.buildHttp2Client;
 import static datadog.communication.http.OkHttpUtils.isPlainHttp;
 
 import datadog.communication.http.HttpRetryPolicy;
-import datadog.communication.otlp.OtlpGrpcRequestBody;
-import datadog.communication.otlp.OtlpPayload;
 import datadog.logging.RatelimitedLogger;
 import datadog.trace.api.config.OtlpConfig.Compression;
-import datadog.trace.common.writer.RemoteApi;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import okhttp3.HttpUrl;
@@ -17,7 +14,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-/** Sends chunks of OTLP data over GRPC. */
+/** OTLP export over gRPC. The HTTP variant is {@link OtlpHttpSender}. */
 public final class OtlpGrpcSender implements OtlpSender {
   private static final Logger LOGGER = LoggerFactory.getLogger(OtlpGrpcSender.class);
   private static final RatelimitedLogger RATELIMITED_LOGGER =
@@ -55,7 +52,7 @@ public OtlpGrpcSender(
   }
 
   @Override
-  public RemoteApi.Response send(OtlpPayload payload) {
+  public OtlpResponse send(OtlpPayload payload) {
     return OtlpSenderSupport.send(client, retryPolicy, makeRequest(payload), RATELIMITED_LOGGER);
   }
 

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

@@ -1,14 +1,11 @@
-package datadog.trace.core.otlp.common;
+package datadog.communication.otlp;
 
 import static datadog.communication.http.OkHttpUtils.buildHttpClient;
 import static datadog.communication.http.OkHttpUtils.isPlainHttp;
 
 import datadog.communication.http.HttpRetryPolicy;
-import datadog.communication.otlp.OtlpHttpRequestBody;
-import datadog.communication.otlp.OtlpPayload;
 import datadog.logging.RatelimitedLogger;
 import datadog.trace.api.config.OtlpConfig.Compression;
-import datadog.trace.common.writer.RemoteApi;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import okhttp3.HttpUrl;
@@ -17,7 +14,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-/** Sends chunks of OTLP data over HTTP. */
+/** OTLP export over HTTP. The gRPC variant is {@link OtlpGrpcSender}. */
 public final class OtlpHttpSender implements OtlpSender {
   private static final Logger LOGGER = LoggerFactory.getLogger(OtlpHttpSender.class);
   private static final RatelimitedLogger RATELIMITED_LOGGER =
@@ -59,7 +56,7 @@ public HttpUrl url() {
   }
 
   @Override
-  public RemoteApi.Response send(OtlpPayload payload) {
+  public OtlpResponse send(OtlpPayload payload) {
     return OtlpSenderSupport.send(client, retryPolicy, makeRequest(payload), RATELIMITED_LOGGER);
   }
 

communication/otlp-exporter/src/main/java/datadog/communication/otlp/OtlpResponse.java

@@ -0,0 +1,74 @@
+package datadog.communication.otlp;
+
+import java.util.Optional;
+import java.util.OptionalInt;
+
+/**
+ * The outcome of a single OTLP export attempt.
+ *
+ * <p>A transport failure or timeout carries only the exception: the payload was not delivered, so
+ * no status code exists. A delivered payload carries the HTTP status code and is successful or not
+ * according to that code. A successful response can still carry a parse exception when its body
+ * could not be read; the payload itself was accepted.
+ */
+public final class OtlpResponse {
+  /** The endpoint accepted the payload and returned no body. */
+  public static OtlpResponse success(final int status) {
+    return new OtlpResponse(true, status, null, null);
+  }
+
+  /** The endpoint accepted the payload and returned a body. */
+  public static OtlpResponse success(final int status, String response) {
+    return new OtlpResponse(true, status, null, response);
+  }
+
+  /** The endpoint accepted the payload, but the response body could not be parsed. */
+  public static OtlpResponse successWithParseError(final int status, final Throwable exception) {
+    return new OtlpResponse(true, status, exception, null);
+  }
+
+  /** The endpoint received the payload but rejected it. */
+  public static OtlpResponse failed(final int status) {
+    return new OtlpResponse(false, status, null, null);
+  }
+
+  /** The endpoint rejected the payload; the body may carry diagnostics. */
+  public static OtlpResponse failed(final int status, String response) {
+    return new OtlpResponse(false, status, null, response);
+  }
+
+  /** The payload never reached the endpoint. */
+  public static OtlpResponse failed(final Throwable exception) {
+    return new OtlpResponse(false, null, exception, null);
+  }
+
+  private final boolean success;
+  private final Integer status;
+  private final Throwable exception;
+  private final String response;
+
+  private OtlpResponse(
+      final boolean success, final Integer status, final Throwable exception, String response) {
+    this.success = success;
+    this.status = status;
+    this.exception = exception;
+    this.response = response;
+  }
+
+  public boolean success() {
+    return success;
+  }
+
+  public OptionalInt status() {
+    return status == null ? OptionalInt.empty() : OptionalInt.of(status);
+  }
+
+  public Optional<Throwable> exception() {
+    return Optional.ofNullable(exception);
+  }
+
+  /** The raw response body, or empty for failed communication attempts (no body received) */
+  public Optional<String> response() {
+    return Optional.ofNullable(response);
+  }
+}

communication/otlp-exporter/src/main/java/datadog/communication/otlp/OtlpSender.java

@@ -0,0 +1,11 @@
+package datadog.communication.otlp;
+
+/**
+ * Sends OTLP payloads to a configured endpoint. Shared by the tracer's OTLP exporters and the
+ * profile exporter so they all reuse one transport implementation.
+ */
+public interface OtlpSender {
+  OtlpResponse send(OtlpPayload payload);
+
+  void shutdown();
+}

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

@@ -1,21 +1,24 @@
-package datadog.trace.core.otlp.common;
+package datadog.communication.otlp;
 
 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 static datadog.communication.otlp.OtlpResponse.failed;
+import static datadog.communication.otlp.OtlpResponse.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(
+  /**
+   * Sends the request and maps every outcome to an {@link OtlpResponse}. Transport failures are
+   * returned instead of thrown, so a broken endpoint degrades to logged warnings and cannot break
+   * the caller's export loop.
+   */
+  static OtlpResponse send(
       OkHttpClient client,
       HttpRetryPolicy.Factory retryPolicy,
       okhttp3.Request request,

communication/otlp-exporter/src/test/java/datadog/communication/otlp/OtlpCanonicalResourceAttributesTest.java

@@ -0,0 +1,75 @@
+package datadog.communication.otlp;
+
+import static datadog.communication.ddagent.TracerVersion.TRACER_VERSION;
+import static java.util.Arrays.asList;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import datadog.trace.api.Config;
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+class OtlpCanonicalResourceAttributesTest {
+
+  @Test
+  void visitsAllAttributesInOrder() {
+    Config config = config("env", "1.0", true, "host");
+
+    assertEquals(
+        asList(
+            "service.name=svc",
+            "deployment.environment.name=env",
+            "service.version=1.0",
+            "host.name=host",
+            "telemetry.sdk.name=datadog",
+            "telemetry.sdk.version=" + TRACER_VERSION,
+            "telemetry.sdk.language=java"),
+        visit(config));
+  }
+
+  @Test
+  void skipsEmptyAndUnreportedAttributes() {
+    Config config = config("", "", false, "host");
+
+    assertEquals(
+        asList(
+            "service.name=svc",
+            "telemetry.sdk.name=datadog",
+            "telemetry.sdk.version=" + TRACER_VERSION,
+            "telemetry.sdk.language=java"),
+        visit(config));
+  }
+
+  @Test
+  void skipsEmptyHostName() {
+    Config config = config("env", "1.0", true, "");
+
+    assertEquals(
+        asList(
+            "service.name=svc",
+            "deployment.environment.name=env",
+            "service.version=1.0",
+            "telemetry.sdk.name=datadog",
+            "telemetry.sdk.version=" + TRACER_VERSION,
+            "telemetry.sdk.language=java"),
+        visit(config));
+  }
+
+  private static Config config(String env, String version, boolean reportHost, String host) {
+    Config config = mock(Config.class);
+    when(config.getServiceName()).thenReturn("svc");
+    when(config.getEnv()).thenReturn(env);
+    when(config.getVersion()).thenReturn(version);
+    when(config.isReportHostName()).thenReturn(reportHost);
+    when(config.getHostName()).thenReturn(host);
+    return config;
+  }
+
+  private static List<String> visit(Config config) {
+    List<String> visited = new ArrayList<>();
+    OtlpCanonicalResourceAttributes.visit(config, (k, v) -> visited.add(k + "=" + v));
+    return visited;
+  }
+}

dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpSenderSupportTest.java → communication/otlp-exporter/src/test/java/datadog/communication/otlp/OtlpSenderSupportTest.java

@@ -1,4 +1,4 @@
-package datadog.trace.core.otlp.common;
+package datadog.communication.otlp;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -12,7 +12,6 @@
 import datadog.communication.http.HttpRetryPolicy;
 import datadog.communication.http.OkHttpUtils;
 import datadog.logging.RatelimitedLogger;
-import datadog.trace.common.writer.RemoteApi;
 import java.io.IOException;
 import okhttp3.MediaType;
 import okhttp3.OkHttpClient;
@@ -39,8 +38,7 @@ void successfulResponseIsReturnedWithoutLogging() throws IOException {
           .when(() -> OkHttpUtils.sendWithRetries(client, retryPolicy, request))
           .thenReturn(response);
 
-      RemoteApi.Response result =
-          OtlpSenderSupport.send(client, retryPolicy, request, ratelimitedLogger);
+      OtlpResponse result = OtlpSenderSupport.send(client, retryPolicy, request, ratelimitedLogger);
 
       assertTrue(result.success());
       assertEquals(200, result.status().getAsInt());
@@ -56,8 +54,7 @@ void unsuccessfulResponseIsReturnedAndLogged() throws IOException {
           .when(() -> OkHttpUtils.sendWithRetries(client, retryPolicy, request))
           .thenReturn(response);
 
-      RemoteApi.Response result =
-          OtlpSenderSupport.send(client, retryPolicy, request, ratelimitedLogger);
+      OtlpResponse result = OtlpSenderSupport.send(client, retryPolicy, request, ratelimitedLogger);
 
       assertFalse(result.success());
       assertEquals(500, result.status().getAsInt());
@@ -73,8 +70,7 @@ void ioExceptionIsReturnedAsFailureAndLogged() throws IOException {
           .when(() -> OkHttpUtils.sendWithRetries(client, retryPolicy, request))
           .thenThrow(exception);
 
-      RemoteApi.Response result =
-          OtlpSenderSupport.send(client, retryPolicy, request, ratelimitedLogger);
+      OtlpResponse result = OtlpSenderSupport.send(client, retryPolicy, request, ratelimitedLogger);
 
       assertFalse(result.success());
       assertTrue(result.exception().isPresent());

communication/otlp-exporter/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker

@@ -0,0 +1 @@
+mock-maker-inline

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

@@ -1,9 +1,10 @@
 package datadog.trace.common.writer;
 
 import datadog.communication.otlp.OtlpPayload;
+import datadog.communication.otlp.OtlpResponse;
+import datadog.communication.otlp.OtlpSender;
 import datadog.trace.api.telemetry.OtlpTelemetry;
 import datadog.trace.core.CoreSpan;
-import datadog.trace.core.otlp.common.OtlpSender;
 import datadog.trace.core.otlp.trace.OtlpTraceCollector;
 import java.util.Collection;
 import java.util.Collections;
@@ -39,7 +40,7 @@ public void flush() {
       OtlpPayload payload = collector.collectTraces();
       if (payload != OtlpPayload.EMPTY) {
         OtlpTelemetry.getInstance().onTracesExportAttempt();
-        RemoteApi.Response response = sender.send(payload);
+        OtlpResponse response = sender.send(payload);
         OtlpTelemetry.getInstance().onTracesExportComplete(response.success());
       }
     } catch (RuntimeException e) { // don't catch severe Errors

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

@@ -6,13 +6,13 @@
 import static datadog.trace.api.ConfigDefaults.DEFAULT_OTLP_TRACES_TIMEOUT;
 
 import datadog.communication.ddagent.DroppingPolicy;
+import datadog.communication.otlp.OtlpGrpcSender;
+import datadog.communication.otlp.OtlpHttpSender;
+import datadog.communication.otlp.OtlpSender;
 import datadog.trace.api.config.OtlpConfig;
 import datadog.trace.common.sampling.SingleSpanSampler;
 import datadog.trace.common.writer.ddagent.Prioritization;
 import datadog.trace.core.monitor.HealthMetrics;
-import datadog.trace.core.otlp.common.OtlpGrpcSender;
-import datadog.trace.core.otlp.common.OtlpHttpSender;
-import datadog.trace.core.otlp.common.OtlpSender;
 import datadog.trace.core.otlp.trace.OtlpTraceCollector;
 import datadog.trace.core.otlp.trace.OtlpTraceJsonCollector;
 import datadog.trace.core.otlp.trace.OtlpTraceProtoCollector;