Create shared DD intake HTTP client

Author: nikita-tkachenko-datadog

communication/src/main/java/datadog/communication/BackendApiFactory.java

@@ -3,13 +3,11 @@
 import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
 import datadog.communication.ddagent.SharedCommunicationObjects;
 import datadog.communication.http.HttpRetryPolicy;
-import datadog.communication.http.OkHttpUtils;
 import datadog.trace.api.Config;
 import datadog.trace.api.intake.Intake;
 import datadog.trace.util.throwable.FatalAgentMisconfigurationError;
 import javax.annotation.Nullable;
 import okhttp3.HttpUrl;
-import okhttp3.OkHttpClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -36,10 +34,13 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni
             "Agentless mode is enabled and api key is not set. Please set application key");
       }
       String traceId = config.getIdGenerationStrategy().generateTraceId().toString();
-      OkHttpClient httpClient =
-          OkHttpUtils.buildHttpClient(
-              agentlessUrl, config.getCiVisibilityBackendApiTimeoutMillis());
-      return new IntakeApi(agentlessUrl, apiKey, traceId, retryPolicyFactory, httpClient, true);
+      return new IntakeApi(
+          agentlessUrl,
+          apiKey,
+          traceId,
+          retryPolicyFactory,
+          sharedCommunicationObjects.getIntakeHttpClient(config),
+          true);
     }
 
     DDAgentFeaturesDiscovery featuresDiscovery =
@@ -55,7 +56,7 @@ public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommuni
           evpProxyUrl,
           subdomain,
           retryPolicyFactory,
-          sharedCommunicationObjects.okHttpClient,
+          sharedCommunicationObjects.agentHttpClient,
           true);
     }
 

communication/src/main/java/datadog/communication/ddagent/SharedCommunicationObjects.java

@@ -27,12 +27,24 @@ public class SharedCommunicationObjects {
   private final List<Runnable> pausedComponents = new ArrayList<>();
   private volatile boolean paused;
 
-  public OkHttpClient okHttpClient;
+  /**
+   * HTTP client for making requests to Datadog agent. Depending on configuration, this client may
+   * use regular HTTP, UDS or named pipe.
+   */
+  public OkHttpClient agentHttpClient;
+
   public HttpUrl agentUrl;
   public Monitoring monitoring;
   private volatile DDAgentFeaturesDiscovery featuresDiscovery;
   private ConfigurationPoller configurationPoller;
 
+  /**
+   * HTTP client for making requests directly to Datadog backend. Unlike {@link #agentHttpClient},
+   * this client is not configured to use UDS or named pipe, and will not force the use of
+   * clear-text HTTP.
+   */
+  private volatile OkHttpClient intakeHttpClient;
+
   public SharedCommunicationObjects() {
     this(false);
   }
@@ -51,12 +63,15 @@ public void createRemaining(Config config) {
         throw new IllegalArgumentException("Bad agent URL: " + config.getAgentUrl());
       }
     }
-    if (okHttpClient == null) {
+    if (agentHttpClient == null) {
       String unixDomainSocket = SocketUtils.discoverApmSocket(config);
       String namedPipe = config.getAgentNamedPipe();
-      okHttpClient =
+      agentHttpClient =
           OkHttpUtils.buildHttpClient(
-              agentUrl, unixDomainSocket, namedPipe, getHttpClientTimeout(config));
+              agentUrl != null && "http".equals(agentUrl.scheme()),
+              unixDomainSocket,
+              namedPipe,
+              getHttpClientTimeout(config));
     }
   }
 
@@ -130,7 +145,7 @@ private ConfigurationPoller createPoller(Config config) {
       configUrlSupplier = new RetryConfigUrlSupplier(this, config);
     }
     return new DefaultConfigurationPoller(
-        config, TRACER_VERSION, containerId, entityId, configUrlSupplier, okHttpClient);
+        config, TRACER_VERSION, containerId, entityId, configUrlSupplier, agentHttpClient);
   }
 
   // for testing
@@ -146,7 +161,7 @@ public DDAgentFeaturesDiscovery featuresDiscovery(Config config) {
           createRemaining(config);
           ret =
               new DDAgentFeaturesDiscovery(
-                  okHttpClient,
+                  agentHttpClient,
                   monitoring,
                   agentUrl,
                   config.isTraceAgentV05Enabled(),
@@ -209,4 +224,19 @@ public String get() {
       return this.configUrl;
     }
   }
+
+  public OkHttpClient getIntakeHttpClient(Config config) {
+    OkHttpClient intakeHttpClient = this.intakeHttpClient;
+    if (intakeHttpClient != null) {
+      return intakeHttpClient;
+    }
+
+    synchronized (this) {
+      if (this.intakeHttpClient == null) {
+        this.intakeHttpClient =
+            OkHttpUtils.buildHttpClient(false, null, null, getHttpClientTimeout(config));
+      }
+      return this.intakeHttpClient;
+    }
+  }
 }

communication/src/main/java/datadog/communication/http/OkHttpUtils.java

@@ -58,19 +58,19 @@ public final class OkHttpUtils {
       SystemProperties.getOrDefault("java.vm.vendor", "unknown");
 
   public static OkHttpClient buildHttpClient(final HttpUrl url, final long timeoutMillis) {
-    return buildHttpClient(url, null, null, timeoutMillis);
+    return buildHttpClient(url != null && "http".equals(url.scheme()), null, null, timeoutMillis);
   }
 
   public static OkHttpClient buildHttpClient(
-      final HttpUrl url,
+      final boolean isHttp,
       final String unixDomainSocketPath,
       final String namedPipe,
       final long timeoutMillis) {
     return buildHttpClient(
         unixDomainSocketPath,
         namedPipe,
         null,
-        url,
+        isHttp,
         null,
         null,
         null,
@@ -95,7 +95,7 @@ public static OkHttpClient buildHttpClient(
         discoverApmSocket(config),
         config.getAgentNamedPipe(),
         dispatcher,
-        url,
+        url != null && "http".equals(url.scheme()),
         retryOnConnectionFailure,
         maxRunningRequests,
         proxyHost,
@@ -111,7 +111,7 @@ private static OkHttpClient buildHttpClient(
       final String unixDomainSocketPath,
       final String namedPipe,
       final Dispatcher dispatcher,
-      final HttpUrl url,
+      final boolean isHttp,
       final Boolean retryOnConnectionFailure,
       final Integer maxRunningRequests,
       final String proxyHost,
@@ -151,7 +151,6 @@ private static OkHttpClient buildHttpClient(
       log.debug("Using NamedPipe as http transport");
     }
 
-    boolean isHttp = url != null && "http".equals(url.scheme());
     if (isHttp) {
       // force clear text when using http to avoid failures for JVMs without TLS
       builder.connectionSpecs(Collections.singletonList(ConnectionSpec.CLEARTEXT));

communication/src/test/groovy/datadog/communication/ddagent/SharedCommunicationsObjectsSpecification.groovy

@@ -24,7 +24,7 @@ class SharedCommunicationsObjectsSpecification extends DDSpecification {
     1 * config.agentTimeout >> 1
     1 * config.agentUnixDomainSocket >> null
     sco.agentUrl as String == 'http://example.com/'
-    sco.okHttpClient != null
+    sco.agentHttpClient != null
     sco.monitoring.is(Monitoring.DISABLED)
 
     when:
@@ -78,7 +78,7 @@ class SharedCommunicationsObjectsSpecification extends DDSpecification {
     DDAgentFeaturesDiscovery agentFeaturesDiscovery = Mock()
 
     sco.agentUrl = url
-    sco.okHttpClient = okHttpClient
+    sco.agentHttpClient = okHttpClient
     sco.monitoring = monitoring
     sco.featuresDiscovery = agentFeaturesDiscovery
 
@@ -88,7 +88,7 @@ class SharedCommunicationsObjectsSpecification extends DDSpecification {
     then:
     0 * _
     sco.agentUrl.is(url)
-    sco.okHttpClient.is(okHttpClient)
+    sco.agentHttpClient.is(okHttpClient)
     sco.monitoring.is(monitoring)
     sco.featuresDiscovery.is(agentFeaturesDiscovery)
   }

dd-java-agent/agent-ci-visibility/src/main/java/datadog/trace/civisibility/CiVisibilityServices.java

@@ -153,7 +153,7 @@ private static CiEnvironment buildCiEnvironment(Config config, SharedCommunicati
       CiEnvironment remoteEnvironment =
           new CiEnvironmentImpl(
               getRemoteEnvironment(
-                  remoteEnvVarsProviderUrl, remoteEnvVarsProviderKey, sco.okHttpClient));
+                  remoteEnvVarsProviderUrl, remoteEnvVarsProviderKey, sco.agentHttpClient));
       return new CompositeCiEnvironment(remoteEnvironment, localEnvironment);
     } else {
       return localEnvironment;

dd-java-agent/appsec/src/jmh/java/datadog/appsec/benchmark/AppSecBenchmark.java

@@ -70,9 +70,9 @@ public void setUp() throws URISyntaxException {
     ss = gw.getSubscriptionService(RequestContextSlot.APPSEC);
     SharedCommunicationObjects sharedCommunicationObjects = new SharedCommunicationObjects();
     sharedCommunicationObjects.monitoring = Monitoring.DISABLED;
-    sharedCommunicationObjects.okHttpClient = new StubOkHttpClient();
+    sharedCommunicationObjects.agentHttpClient = new StubOkHttpClient();
     sharedCommunicationObjects.setFeaturesDiscovery(
-        new StubDDAgentFeaturesDiscovery(sharedCommunicationObjects.okHttpClient));
+        new StubDDAgentFeaturesDiscovery(sharedCommunicationObjects.agentHttpClient));
 
     AppSecSystem.start(ss, sharedCommunicationObjects);
     uri = new URIDefaultDataAdapter(new URI("http://localhost:8080/test"));

dd-java-agent/appsec/src/test/groovy/com/datadog/appsec/AppSecSystemSpecification.groovy

@@ -182,7 +182,7 @@ class AppSecSystemSpecification extends DDSpecification {
           poller
         }
       }
-    sco.okHttpClient = Stub(OkHttpClient)
+    sco.agentHttpClient = Stub(OkHttpClient)
     sco.monitoring = Mock(Monitoring)
     sco.featuresDiscovery = Stub(DDAgentFeaturesDiscovery)
     sco

dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/InstrumentationSpecification.groovy

@@ -371,7 +371,7 @@ abstract class InstrumentationSpecification extends DDSpecification implements A
 
       // emit traces to the APM Test-Agent for Cross-Tracer Testing Trace Checks
       HttpUrl agentUrl = HttpUrl.get("http://" + agentHost + ":" + DEFAULT_TRACE_AGENT_PORT)
-      OkHttpClient client = buildHttpClient(agentUrl, null, null, TimeUnit.SECONDS.toMillis(DEFAULT_AGENT_TIMEOUT))
+      OkHttpClient client = buildHttpClient(true, null, null, TimeUnit.SECONDS.toMillis(DEFAULT_AGENT_TIMEOUT))
       DDAgentFeaturesDiscovery featureDiscovery = new DDAgentFeaturesDiscovery(client, Monitoring.DISABLED, agentUrl, Config.get().isTraceAgentV05Enabled(), Config.get().isTracerMetricsEnabled())
       TEST_AGENT_API = new DDAgentApi(client, agentUrl, featureDiscovery, Monitoring.DISABLED, Config.get().isTracerMetricsEnabled())
       TEST_AGENT_WRITER = DDAgentWriter.builder().agentApi(TEST_AGENT_API).build()

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

@@ -113,7 +113,7 @@ public ConflatingMetricsAggregator(
         sharedCommunicationObjects.featuresDiscovery(config),
         healthMetrics,
         new OkHttpSink(
-            sharedCommunicationObjects.okHttpClient,
+            sharedCommunicationObjects.agentHttpClient,
             sharedCommunicationObjects.agentUrl.toString(),
             V6_METRICS_ENDPOINT,
             config.isTracerMetricsBufferingEnabled(),

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

@@ -138,7 +138,7 @@ public DDAgentWriter build() {
       final HttpUrl agentUrl = HttpUrl.get("http://" + agentHost + ":" + traceAgentPort);
       final OkHttpClient client =
           null == featureDiscovery || null == agentApi
-              ? buildHttpClient(agentUrl, unixDomainSocket, namedPipe, timeoutMillis)
+              ? buildHttpClient(true, unixDomainSocket, namedPipe, timeoutMillis)
               : null;
       if (null == featureDiscovery) {
         featureDiscovery =