Merge branch 'master' into vandonr/dsm

.circleci/config.continue.yml.j2

@@ -1092,6 +1092,20 @@ build_test_jobs: &build_test_jobs
       maxWorkers: 4
       testJvm: "8"
 
+  - xlarge_tests:
+      requires:
+        - ok_to_test
+        - build_latestdep
+      name: test_17_inst_latest
+      gradleTarget: ":instrumentationLatestDepTest"
+      gradleParameters: "-PskipFlakyTests"
+      triggeredBy: *instrumentation_modules
+      stage: instrumentation
+      cacheType: latestdep
+      parallelism: 4
+      maxWorkers: 4
+      testJvm: "17"
+
 {% if flaky %}
   - tests:
       requires:

.github/CODEOWNERS

@@ -20,6 +20,7 @@ dd-java-agent/instrumentation/jacoco/      @DataDog/ci-app-libraries-java
 dd-java-agent/instrumentation/junit-4.10/  @DataDog/ci-app-libraries-java
 dd-java-agent/instrumentation/junit-5.3/   @DataDog/ci-app-libraries-java
 dd-java-agent/instrumentation/karate/      @DataDog/ci-app-libraries-java
+dd-java-agent/instrumentation/scalatest/   @DataDog/ci-app-libraries-java
 dd-java-agent/instrumentation/testng/      @DataDog/ci-app-libraries-java
 dd-java-agent/instrumentation/gradle/      @DataDog/ci-app-libraries-java
 dd-java-agent/instrumentation/maven-3.2.1/ @DataDog/ci-app-libraries-java

.gitlab/benchmarks.yml

@@ -30,3 +30,40 @@ benchmarks:
 
     KUBERNETES_SERVICE_ACCOUNT_OVERWRITE: dd-trace-java
     FF_USE_LEGACY_KUBERNETES_EXECUTION_STRATEGY: "true"
+
+.dsm-kafka-benchmarks:
+  stage: benchmarks
+  rules:
+    - if: $CI_PIPELINE_SOURCE != "schedule"
+      changes:
+        paths:
+          - dd-java-agent/instrumentation/kafka*/**/*
+        compare_to: "master"
+      when: on_success
+    - when: manual
+  tags: ["runner:apm-k8s-tweaked-metal"]
+  interruptible: true
+  timeout: 1h
+  image: 486234852809.dkr.ecr.us-east-1.amazonaws.com/ci/benchmarking-platform:java-dsm-kafka
+  script:
+    - git clone --branch java/kafka-dsm-overhead https://gitlab-ci-token:${CI_JOB_TOKEN}@gitlab.ddbuild.io/DataDog/benchmarking-platform.git platform && cd platform
+    - ./steps/run-benchmarks.sh
+  artifacts:
+    name: "artifacts"
+    when: always
+    paths:
+      - platform/artifacts/
+    expire_in: 3 months
+  variables:
+    FF_USE_LEGACY_KUBERNETES_EXECUTION_STRATEGY: "true"
+
+dsm-kafka-producer-benchmark:
+  extends: .dsm-kafka-benchmarks
+  variables:
+    BP_KAFKA_SCENARIO_DIR: producer-benchmark
+
+dsm-kafka-consumer-benchmark:
+  extends: .dsm-kafka-benchmarks
+  variables:
+    BP_KAFKA_SCENARIO_DIR: consumer-benchmark
+

build.gradle

@@ -5,7 +5,7 @@ buildscript {
 
   configurations.all {
     resolutionStrategy.dependencySubstitution {
-      substitute module("com.jcraft:jsch") using module("com.github.mwiede:jsch:0.2.6") because "jcraft jsch has been unmaintained for years."
+      substitute module("com.jcraft:jsch") using module("com.github.mwiede:jsch:0.2.14") because "jcraft jsch has been unmaintained for years."
     }
   }
 }

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

@@ -46,6 +46,7 @@ public class DDAgentFeaturesDiscovery implements DroppingPolicy {
   public static final String DATADOG_AGENT_STATE = "Datadog-Agent-State";
 
   public static final String DEBUGGER_ENDPOINT = "debugger/v1/input";
+  public static final String DEBUGGER_DIAGNOSTICS_ENDPOINT = "debugger/v1/diagnostics";
 
   public static final String TELEMETRY_PROXY_ENDPOINT = "telemetry/proxy/";
 
@@ -70,6 +71,7 @@ public class DDAgentFeaturesDiscovery implements DroppingPolicy {
   private volatile String state;
   private volatile String configEndpoint;
   private volatile String debuggerEndpoint;
+  private volatile String debuggerDiagnosticsEndpoint;
   private volatile String evpProxyEndpoint;
   private volatile String version;
   private volatile String telemetryProxyEndpoint;
@@ -100,6 +102,7 @@ private void reset() {
     state = null;
     configEndpoint = null;
     debuggerEndpoint = null;
+    debuggerDiagnosticsEndpoint = null;
     dataStreamsEndpoint = null;
     evpProxyEndpoint = null;
     version = null;
@@ -211,7 +214,7 @@ private boolean processInfoResponse(String response) {
       String foundMetricsEndpoint = null;
       if (metricsEnabled) {
         for (String endpoint : metricsEndpoints) {
-          if (endpoints.contains(endpoint) || endpoints.contains("/" + endpoint)) {
+          if (containsEndpoint(endpoints, endpoint)) {
             foundMetricsEndpoint = endpoint;
             break;
           }
@@ -222,39 +225,42 @@ private boolean processInfoResponse(String response) {
       metricsEndpoint = foundMetricsEndpoint;
 
       for (String endpoint : traceEndpoints) {
-        if (endpoints.contains(endpoint) || endpoints.contains("/" + endpoint)) {
+        if (containsEndpoint(endpoints, endpoint)) {
           traceEndpoint = endpoint;
           break;
         }
       }
 
       for (String endpoint : configEndpoints) {
-        if (endpoints.contains(endpoint) || endpoints.contains("/" + endpoint)) {
+        if (containsEndpoint(endpoints, endpoint)) {
           configEndpoint = endpoint;
           break;
         }
       }
 
-      if (endpoints.contains(DEBUGGER_ENDPOINT) || endpoints.contains("/" + DEBUGGER_ENDPOINT)) {
+      if (containsEndpoint(endpoints, DEBUGGER_ENDPOINT)) {
         debuggerEndpoint = DEBUGGER_ENDPOINT;
       }
+      if (containsEndpoint(endpoints, DEBUGGER_DIAGNOSTICS_ENDPOINT)) {
+        debuggerDiagnosticsEndpoint = DEBUGGER_DIAGNOSTICS_ENDPOINT;
+      }
 
       for (String endpoint : dataStreamsEndpoints) {
-        if (endpoints.contains(endpoint) || endpoints.contains("/" + endpoint)) {
+        if (containsEndpoint(endpoints, endpoint)) {
           dataStreamsEndpoint = endpoint;
           break;
         }
       }
 
       for (String endpoint : evpProxyEndpoints) {
-        if (endpoints.contains(endpoint) || endpoints.contains("/" + endpoint)) {
+        if (containsEndpoint(endpoints, endpoint)) {
           evpProxyEndpoint = endpoint;
           break;
         }
       }
 
       for (String endpoint : telemetryProxyEndpoints) {
-        if (endpoints.contains(endpoint) || endpoints.contains("/" + endpoint)) {
+        if (containsEndpoint(endpoints, endpoint)) {
           telemetryProxyEndpoint = endpoint;
           break;
         }
@@ -281,6 +287,10 @@ private boolean processInfoResponse(String response) {
     return false;
   }
 
+  private static boolean containsEndpoint(Set<String> endpoints, String endpoint) {
+    return endpoints.contains(endpoint) || endpoints.contains("/" + endpoint);
+  }
+
   @SuppressWarnings("unchecked")
   private static void discoverStatsDPort(final Map<String, Object> info) {
     try {
@@ -309,6 +319,10 @@ public boolean supportsDebugger() {
     return debuggerEndpoint != null;
   }
 
+  public boolean supportsDebuggerDiagnostics() {
+    return debuggerDiagnosticsEndpoint != null;
+  }
+
   boolean supportsDropping() {
     return supportsDropping;
   }

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

@@ -59,6 +59,7 @@ class DDAgentFeaturesDiscoveryTest extends DDSpecification {
     features.state() == INFO_STATE
     features.getConfigEndpoint() == V7_CONFIG_ENDPOINT
     features.supportsDebugger()
+    features.supportsDebuggerDiagnostics()
     features.supportsEvpProxy()
     features.getVersion() == "0.99.0"
     !features.supportsLongRunning()
@@ -87,6 +88,7 @@ class DDAgentFeaturesDiscoveryTest extends DDSpecification {
     features.state() == INFO_STATE
     features.getConfigEndpoint() == V7_CONFIG_ENDPOINT
     features.supportsDebugger()
+    features.supportsDebuggerDiagnostics()
     features.supportsEvpProxy()
     features.getVersion() == "0.99.0"
     !features.supportsLongRunning()

communication/src/test/resources/agent-features/agent-info.json

@@ -14,6 +14,7 @@
     "/evp_proxy/v1/",
     "/evp_proxy/v2/",
     "/debugger/v1/input",
+    "/debugger/v1/diagnostics",
     "/v0.7/config"
   ],
   "feature_flags": [

dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java

@@ -220,7 +220,7 @@ public static void start(final Instrumentation inst, final URL agentJarURL, Stri
         // multiple times
         // If early profiling is enabled then this call will start profiling.
         // If early profiling is disabled then later call will do this.
-        startProfilingAgent(true);
+        startProfilingAgent(true, inst);
       } else {
         log.debug("Oracle JDK 8 detected. Delaying profiler initialization.");
         // Profiling can not run early on Oracle JDK 8 because it will cause JFR initialization
@@ -230,7 +230,7 @@ public static void start(final Instrumentation inst, final URL agentJarURL, Stri
             new Runnable() {
               @Override
               public void run() {
-                startProfilingAgent(false);
+                startProfilingAgent(false, inst);
               }
             };
       }
@@ -309,9 +309,9 @@ public void run() {
 
       if (delayOkHttp) {
         log.debug("Custom logger detected. Delaying Profiling initialization.");
-        registerLogManagerCallback(new StartProfilingAgentCallback());
+        registerLogManagerCallback(new StartProfilingAgentCallback(inst));
       } else {
-        startProfilingAgent(false);
+        startProfilingAgent(false, inst);
         // only enable instrumentation based profilers when we know JFR is ready
         InstrumentationBasedProfiling.enableInstrumentationBasedProfiling();
       }
@@ -468,7 +468,7 @@ public void execute() {
       installDatadogTracer(scoClass, sco);
       maybeStartAppSec(scoClass, sco);
       maybeStartIast(scoClass, sco);
-      maybeStartCiVisibility(scoClass, sco);
+      maybeStartCiVisibility(instrumentation, scoClass, sco);
       // start debugger before remote config to subscribe to it before starting to poll
       maybeStartDebugger(instrumentation, scoClass, sco);
       maybeStartRemoteConfig(scoClass, sco);
@@ -480,14 +480,20 @@ public void execute() {
   }
 
   protected static class StartProfilingAgentCallback extends ClassLoadCallBack {
+    private final Instrumentation inst;
+
+    protected StartProfilingAgentCallback(Instrumentation inst) {
+      this.inst = inst;
+    }
+
     @Override
     public AgentThread agentThread() {
       return PROFILER_STARTUP;
     }
 
     @Override
     public void execute() {
-      startProfilingAgent(false);
+      startProfilingAgent(false, inst);
       // only enable instrumentation based profilers when we know JFR is ready
       InstrumentationBasedProfiling.enableInstrumentationBasedProfiling();
     }
@@ -756,16 +762,16 @@ private static void startIast(SubscriptionService ss, Class<?> scoClass, Object
     }
   }
 
-  private static void maybeStartCiVisibility(Class<?> scoClass, Object sco) {
+  private static void maybeStartCiVisibility(Instrumentation inst, Class<?> scoClass, Object sco) {
     if (ciVisibilityEnabled) {
       StaticEventLogger.begin("CI Visibility");
 
       try {
         final Class<?> ciVisibilitySysClass =
             AGENT_CLASSLOADER.loadClass("datadog.trace.civisibility.CiVisibilitySystem");
         final Method ciVisibilityInstallerMethod =
-            ciVisibilitySysClass.getMethod("start", scoClass);
-        ciVisibilityInstallerMethod.invoke(null, sco);
+            ciVisibilitySysClass.getMethod("start", Instrumentation.class, scoClass);
+        ciVisibilityInstallerMethod.invoke(null, inst, sco);
       } catch (final Throwable e) {
         log.warn("Not starting CI Visibility subsystem", e);
       }
@@ -881,7 +887,7 @@ private static ProfilingContextIntegration createProfilingContextIntegration() {
     return ProfilingContextIntegration.NoOp.INSTANCE;
   }
 
-  private static void startProfilingAgent(final boolean isStartingFirst) {
+  private static void startProfilingAgent(final boolean isStartingFirst, Instrumentation inst) {
     StaticEventLogger.begin("ProfilingAgent");
 
     if (isAwsLambdaRuntime()) {
@@ -895,8 +901,9 @@ private static void startProfilingAgent(final boolean isStartingFirst) {
       final Class<?> profilingAgentClass =
           AGENT_CLASSLOADER.loadClass("com.datadog.profiling.agent.ProfilingAgent");
       final Method profilingInstallerMethod =
-          profilingAgentClass.getMethod("run", Boolean.TYPE, ClassLoader.class);
-      profilingInstallerMethod.invoke(null, isStartingFirst, AGENT_CLASSLOADER);
+          profilingAgentClass.getMethod(
+              "run", Boolean.TYPE, ClassLoader.class, Instrumentation.class);
+      profilingInstallerMethod.invoke(null, isStartingFirst, AGENT_CLASSLOADER, inst);
       /*
        * Install the tracer hooks only when not using 'early start'.
        * The 'early start' is happening so early that most of the infrastructure has not been set up yet.

dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/ContextStore.java

@@ -16,10 +16,26 @@ public interface ContextStore<K, C> {
    *
    * @param <C> context type
    */
-  interface Factory<C> {
+  interface Factory<C> extends KeyAwareFactory<Object, C> {
 
     /** @return new context instance */
     C create();
+
+    default C create(Object key) {
+      return create();
+    }
+  }
+
+  /**
+   * Factory interface to create context instances using context key instances
+   *
+   * @param <K> context key type
+   * @param <C> context value type
+   */
+  interface KeyAwareFactory<K, C> {
+
+    /** @return new context instance */
+    C create(K key);
   }
 
   /**
@@ -57,6 +73,16 @@ interface Factory<C> {
    */
   C putIfAbsent(K key, Factory<C> contextFactory);
 
+  /**
+   * Put new context instance if key is absent. Uses context factory to avoid creating objects if
+   * not needed.
+   *
+   * @param key key to use
+   * @param contextFactory factory instance to produce new context object
+   * @return old instance if it was present, or new instance
+   */
+  C computeIfAbsent(K key, KeyAwareFactory<? super K, C> contextFactory);
+
   /**
    * Removes the existing value for key and return it.
    *

dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/FieldBackedContextStore.java

@@ -51,21 +51,27 @@ public Object putIfAbsent(final Object key, final Object context) {
 
   @Override
   public Object putIfAbsent(final Object key, final Factory<Object> contextFactory) {
+    return computeIfAbsent(key, contextFactory);
+  }
+
+  @Override
+  public Object computeIfAbsent(
+      Object key, KeyAwareFactory<? super Object, Object> contextFactory) {
     if (key instanceof FieldBackedContextAccessor) {
       final FieldBackedContextAccessor accessor = (FieldBackedContextAccessor) key;
       Object existingContext = accessor.$get$__datadogContext$(storeId);
       if (null == existingContext) {
         synchronized (accessor) {
           existingContext = accessor.$get$__datadogContext$(storeId);
           if (null == existingContext) {
-            existingContext = contextFactory.create();
+            existingContext = contextFactory.create(key);
             accessor.$put$__datadogContext$(storeId, existingContext);
           }
         }
       }
       return existingContext;
     } else {
-      return weakStore().putIfAbsent(key, contextFactory);
+      return weakStore().computeIfAbsent(key, contextFactory);
     }
   }